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

// Thread sessions (row 121, #1615): the launcher starts a thread instance's
// session when its thread has a pending message or was opened, stops it once
// the thread is idle, and gives every run of a thread the thread's own store.
// Real runners in the bwrap sandbox; a stub adapter keeps a history in the
// store and answers with what it found there.
//
// The host is in process: startBroker on bootConfig's config, with the ops
// the broker child forwards (packages/bus/src/process.mjs). The real host
// proves a human CLI process for every human verb, which a test run under an
// agent can't, so the test's human capability comes from bindHuman directly.

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { existsSync, mkdirSync, readdirSync, 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 { Client } from "../../bus/src/client.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, SESSION_CREDENTIALS } from "../../seat/src/session.mjs";
import { bootConfig, loadSystem } from "../src/config.mjs";
import { startTimeOf } from "../src/host.mjs";
import { createLauncher } from "../src/launcher.mjs";
import { fixture, tmp } from "./helpers.mjs";

const ADAPTER = fileURLToPath(new URL("./fixtures/thread-adapter.mjs", import.meta.url));
const skip = sandboxProblem() ?? false;

async function until(fn, ms = 20000) {
  const end = Date.now() + ms;
  while (Date.now() < end) {
    const v = await fn();
    if (v) return v;
    await new Promise((r) => setTimeout(r, 50));
  }
  throw new Error("timed out waiting");
}

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; sonnet is its model's limit.
function prepare(t, { sonnet = 2, idle = 1 } = {}) {
  const root = tmp(t, "mosaic-thread-");
  const f = fixture(root, "acme", (doc) => {
    doc.roles.chief.vars = { harness: "pi", model: "claude-opus-5-5" };
    doc.roles.assistant = { definition: "assistant", vars: { harness: "pi", model: "claude-sonnet-5-5" } };
    doc.launch = launchDoc("chief", ["coder", "assistant"], { opus: 1, sonnet });
    doc.launch.threadIdleSeconds = idle;
    return doc;
  });
  for (const k of SESSION_CREDENTIALS.flat()) delete f.env[k];
  mkdirSync(f.dataRoot, { mode: 0o700 });
  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 });
  return { root, f, adapter };
}

// A host and its launcher on the prepared root. close() runs the launcher's
// close hook, then closes the broker, as the real host does. `ops` replaces
// launch ops, given the runtime.
async function start(t, { f, adapter }, { tickMs = 100, ops: replace } = {}) {
  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);
  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) => runtime.threadsState(m.business),
    threadIdle: (m) => runtime.threadIdle(m.record),
    ...replace?.(runtime),
  };
  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,
    adapters: { pi: adapter, "claude-code": adapter },
    sessionDefaults: { pollInterval: 100 },
    versions: { pi: "0.85.1", claude: null },
    readOnly: [ADAPTER],
    threadTickMs: tickMs,
    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 });
  const call = (verb, args = {}) => runtime.broker.request(human, { verb, args });
  return { host, runtime, launcher, call, 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 threadRuns = (f, thread) => readSessions(f.dataRoot).filter((s) => s.thread === thread);

// Send to a thread and wait for the assistant's RESULT reply in that thread.
async function ask(call, thread, body) {
  const sent = call("thread.send", { thread, body });
  const reply = await until(() => call("thread.messages", { thread }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT"));
  assert.equal(reply.from_role, "assistant");
  assert.equal(reply.to_role, "human");
  return { sent, reply, saw: JSON.parse(reply.body.match(/THREAD (\{.*\})/)[1]) };
}

// The run stops on its own once the thread is idle: no session, no holder.
async function idleStop(f, host, thread, run) {
  await until(() => launchLog(f).find((e) => e.event === "end" && e.run === run));
  assert.ok(launchLog(f).find((e) => e.event === "thread-idle" && e.run === run && e.thread === thread));
  assert.deepEqual(threadRuns(f, thread), []);
  assert.equal(host.runtime.threadsState("acme").find((x) => x.thread === thread).holder, null);
}

test("REQ-THREAD-2 REQ-LAUNCH-1 REQ-MSG-1 a thread message starts the thread's session, which stops when idle; its history survives the stop and a restart and no other thread sees it", { skip, timeout: 90000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  let host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const B = host.call("thread.create", { project, name: "beta" }).id;

  // Nothing runs for a thread with nothing pending.
  await host.launcher.tick();
  assert.deepEqual(readSessions(f.dataRoot), []);

  const first = await ask(host.call, A, "first for A");
  assert.deepEqual(first.saw, { body: "first for A", before: [], transcripts: [] });
  const [run1] = launchLog(f).filter((e) => e.event === "launch" && e.thread === A).map((e) => e.run);
  assert.ok(run1);
  const store = join(f.dataRoot, "threads", "acme", A);
  assert.equal(readFileSync(join(store, "session", "history"), "utf8"), "first for A\n");
  assert.deepEqual(readdirSync(join(store, "claude-projects")), ["turn-1.jsonl"]);
  assert.equal(existsSync(join(f.dataRoot, "launches", "acme", run1, "session")), false, "a thread run keeps no per-run session directory");
  await idleStop(f, host, A, run1);

  // A second run of the same thread finds the first run's history.
  const second = await ask(host.call, A, "second for A");
  assert.deepEqual(second.saw, { body: "second for A", before: ["first for A"], transcripts: ["turn-1.jsonl"] });
  const run2 = launchLog(f).filter((e) => e.event === "launch" && e.thread === A).at(-1).run;
  assert.notEqual(run2, run1);
  await idleStop(f, host, A, run2);

  // A restart: a new host and launcher on the same data root.
  await host.close();
  host = await start(t, prepared);
  const third = await ask(host.call, A, "third for A");
  assert.deepEqual(third.saw, { body: "third for A", before: ["first for A", "second for A"], transcripts: ["turn-1.jsonl", "turn-2.jsonl"] });

  // Thread B has its own store, empty, and A's messages aren't in it.
  const other = await ask(host.call, B, "for B");
  assert.deepEqual(other.saw, { body: "for B", before: [], transcripts: [] });
  assert.ok(!host.call("thread.messages", { thread: B }).some((m) => /for A/.test(m.body)));
  assert.ok(!host.call("thread.messages", { thread: A }).some((m) => /for B/.test(m.body)));

  // Opening a thread starts its session with nothing pending; it idles out too.
  for (const r of readSessions(f.dataRoot)) await idleStop(f, host, r.thread, r.run);
  host.call("thread.open", { thread: B });
  const opened = await until(() => threadRuns(f, B)[0]);
  await idleStop(f, host, B, opened.run);
});

test("REQ-THREAD-2 REQ-LAUNCH-1 a running thread counts against its model's limit and an idle-stopped one doesn't", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t, { sonnet: 1 });
  const { f } = prepared;
  const host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const B = host.call("thread.create", { project, name: "beta" }).id;

  const toA = host.call("thread.send", { thread: A, body: "for A" });
  const toB = host.call("thread.send", { thread: B, body: "for B" });
  const refused = await until(() => launchLog(f).find((e) => e.event === "refused" && e.thread === B));
  assert.equal(refused.code, "capacity-full");
  assert.equal(refused.instance, "assistant");
  const replyB = await until(() => host.call("thread.messages", { thread: B }).find((m) => m.in_reply_to === toB.id), 30000);
  assert.ok(host.call("thread.messages", { thread: A }).find((m) => m.in_reply_to === toA.id));

  // B started only after A's run ended, and its refusal was logged once.
  const entries = launchLog(f);
  const endA = entries.findIndex((e) => e.event === "end" && e.run === entries.find((x) => x.event === "launch" && x.thread === A).run);
  const launchB = entries.findIndex((e) => e.event === "launch" && e.thread === B);
  assert.ok(endA >= 0 && launchB > endA, JSON.stringify(entries.map((e) => [e.event, e.thread, e.code])));
  assert.equal(entries.filter((e) => e.event === "refused" && e.thread === B).length, 1);
  assert.match(replyB.body, /THREAD/);
});

test("REQ-THREAD-2 REQ-LAUNCH-1 a turn longer than the idle time isn't stopped, and the thread idles out after it", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  // The stub's "slow" turn takes 3 s and the idle time is 1 s. The message is
  // unread for the whole turn.
  const sent = host.call("thread.send", { thread: A, body: "slow" });
  const reply = () => host.call("thread.messages", { thread: A }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT");
  await until(() => reply() || launchLog(f).find((e) => e.event === "thread-idle"), 30000);
  assert.ok(reply(), JSON.stringify(launchLog(f)));
  await idleStop(f, host, A, launchLog(f).find((e) => e.event === "launch" && e.thread === A).run);
});

// Dewey's F3 (28591): the tick decides from a snapshot. A message sent after it
// keeps the session, which answers it, instead of going to a run being stopped.
test("REQ-THREAD-2 REQ-LAUNCH-1 a message sent after the idle snapshot keeps the session, which answers it", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  let race = null;
  // Ticks only when the test calls tick(). The snapshot is taken before the race's send.
  const host = await start(t, prepared, {
    tickMs: 3600000,
    ops: (runtime) => ({
      threadsState: (m) => {
        const snapshot = runtime.threadsState(m.business);
        race?.();
        race = null;
        return snapshot;
      },
    }),
  });
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const reply = (sent) => host.call("thread.messages", { thread: A }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT");
  const first = host.call("thread.send", { thread: A, body: "first for A" });
  await host.launcher.tick();
  await until(() => reply(first));
  const [launch] = launchLog(f).filter((e) => e.event === "launch" && e.thread === A);
  // Past the idle time, so the snapshot says stop.
  await new Promise((r) => setTimeout(r, 1200));
  let late;
  race = () => (late = host.call("thread.send", { thread: A, body: "late for A" }));
  await host.launcher.tick();
  assert.ok(late, "the race ran");
  assert.equal(launchLog(f).find((e) => e.event === "thread-idle"), undefined, JSON.stringify(launchLog(f)));
  await until(() => reply(late));
  assert.deepEqual(launchLog(f).filter((e) => e.event === "launch").map((e) => e.run), [launch.run]);
  // With nothing waiting, the next tick past the idle time stops it.
  await new Promise((r) => setTimeout(r, 1200));
  await host.launcher.tick();
  await idleStop(f, host, A, launch.run);
});

// A thread held by a run the launcher has no session for, as after a restart
// before recovery. The hold is the bus's, so the launcher starts nothing.
test("REQ-ROLE-2 REQ-LAUNCH-1 a thread held by a run the launcher didn't start gets no second session", { skip, timeout: 30000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const held = host.runtime.bindLaunch({ business: "acme", role: "assistant", run: "held-by-hand", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid), thread: A });
  host.runtime.broker.request(held.cap, { verb: "role.claim" });
  assert.equal(host.runtime.threadsState("acme").find((x) => x.thread === A).holder, "held-by-hand");
  host.call("thread.send", { thread: A, body: "for A" });
  for (let i = 0; i < 3; i++) await host.launcher.tick();
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.deepEqual(launchLog(f), []);
});

// A run that ends before it takes the message it was started for. The next
// start waits for a new message or open, so a run that dies at once doesn't
// start again on every tick.
test("REQ-LAUNCH-1 a run that ends without taking its thread's message isn't started again for that message", { skip, timeout: 30000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  // Ticks only when the test calls tick().
  const host = await start(t, prepared, { tickMs: 3600000 });
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  const A = host.call("thread.create", { project, name: "alpha" }).id;
  const held = host.runtime.bindLaunch({ business: "acme", role: "assistant", run: "held-by-hand", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid), thread: A });
  host.call("thread.send", { thread: A, body: "for A" });
  await host.launcher.tick();
  // tick() returns once the run is bound and its capability is queued on its
  // stdin. This claim is synchronous, so it comes before the run's own.
  host.runtime.broker.request(held.cap, { verb: "role.claim" });
  const [launch] = launchLog(f).filter((e) => e.event === "launch" && e.thread === A);
  const end = await until(() => launchLog(f).find((e) => e.event === "end" && e.run === launch.run));
  assert.equal(end.exitCode, 21, "the run's claim was refused");
  host.runtime.broker.request(held.cap, { verb: "role.release" });
  const state = host.runtime.threadsState("acme").find((x) => x.thread === A);
  assert.deepEqual([state.holder, state.pending], [null, 1]);
  for (let i = 0; i < 3; i++) await host.launcher.tick();
  assert.equal(launchLog(f).filter((e) => e.event === "launch").length, 1);
  assert.deepEqual(readSessions(f.dataRoot), []);
});

// A thread id that can't be a path segment, made by hand in the store before
// the host starts, and a thread of a non-thread instance. Neither launches.
test("REQ-THREAD-2 REQ-LAUNCH-1 a thread id that isn't a safe segment, or a thread of a non-thread instance, launches nothing", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const store = new Store(f.dataRoot);
  const at = () => new Date().toISOString();
  store.run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)", "p-hand", at(), "acme", "hand made", "stack", "jason");
  for (const [id, instance] of [["../evil", "assistant"], ["coder-thread", "coder"]]) {
    store.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", id, at(), "acme", "p-hand", instance, "jason", id);
    store.run("INSERT INTO thread_opens(at,thread,by) VALUES(?,?,?)", at(), id, "jason");
  }
  store.close();

  const host = await start(t, prepared);
  const refusals = await until(() => {
    const r = launchLog(f).filter((e) => e.event === "refused");
    return r.length === 2 && r;
  });
  assert.deepEqual(refusals.map((e) => [e.thread, e.instance, e.code]).sort(), [["../evil", "assistant", "thread-store-refused"], ["coder-thread", "coder", "thread-not-allowed"]]);
  assert.match(refusals.find((e) => e.thread === "../evil").why, /thread "..\/evil" isn't an id/);
  // More ticks log nothing more and start nothing.
  for (let i = 0; i < 3; i++) await host.launcher.tick();
  assert.equal(launchLog(f).length, 2);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.equal(existsSync(join(f.dataRoot, "threads")), false);
  assert.equal(existsSync(join(f.dataRoot, "evil")), false);
  assert.equal(existsSync(join(f.dataRoot, "launches", "acme")), false, "no run directory");
  assert.equal(existsSync(join(f.dataRoot, "workspaces")), false, "no workspace");
});

test("REQ-THREAD-1 REQ-LAUNCH-1 the chief can't launch the assistant outside a thread", { skip, timeout: 30000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const chief = await host.host.bindLaunch({ business: "acme", role: "chief", run: "chief-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
  await new Client({ path: host.host.path, cap: chief.cap }).call("role.claim");
  assert.deepEqual(await host.launcher.fromSession(chief.cap, "assistant"), { ok: false, error: "thread-required" });
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.deepEqual(launchLog(f).map((e) => [e.event, e.instance, e.code]), [["refused", "assistant", "thread-required"]]);
});