stack/packages/cli/tests/launcher-final-race-probe.test.mjs

import cp from 'node:child_process';
import { syncBuiltinESMExports } from 'node:module';
import { DatabaseSync } from 'node:sqlite';
import { requestHostOps } from './request-host-ops.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, readdirSync, 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 { 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 } 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 { fakeHarnesses, fixture, tmp, withoutHostCredentials } from './helpers.mjs';
import { mappingEntry } from '../../library/tests/helpers.mjs';
// Row 264a (#1769): a thread made by hand gets the default session the 3k-3l migration gives a
// legacy thread, with the thread's id.
const DEFAULT_SESSION = [
  "INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3l' FROM threads WHERE id=?",
  "INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3l' FROM threads WHERE id=?",
];

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, edit = (doc) => doc } = {}) {
  const root = tmp(t, "mosaic-thread-");
  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 });
    doc.launch.threadIdleSeconds = idle;
    return edit(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 });
  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 = {
    ...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),
    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),
    ...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,
    makeLoader: await fakeHarnesses(t, { pi: adapter, "claude-code": adapter }),
    sessionDefaults: { pollInterval: 100 },
    harnessVersion: "0.85.1",
    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('archive during final project admission must still prevent thread runner spawn', {skip,timeout:90000},async t=>{
 const prepared=prepare(t),{f}=prepared;
 let project,reads=0,spawned=0,archived=false;
 const h=await start(t,prepared,{tickMs:60000,ops:runtime=>({
  'request.admission':m=>{
   const answer=runtime.admission(m);
   if(project&&m.project==='stack'&&++reads===2){
    const cap=runtime.broker.bindHuman({business:'acme',human:f.doc.human,via:'cli',outsideAgent:true});
    runtime.broker.request(cap,{verb:'project.archive',args:{project}});archived=true;
   }
   return answer;
  }
 })});
 project=h.call('project.create',{name:'intake without repo'}).id;
 const thread=h.call('thread.create',{project,name:'queued'}).id;
 h.call('thread.open',{thread});
 const original=cp.spawn;
 cp.spawn=(command,args,opts)=>{
  if(opts?.cwd?.startsWith(join(f.dataRoot,'launches')))spawned++;
  return original(command,args,opts);
 };
 syncBuiltinESMExports();
 try {await h.launcher.tick();}finally{cp.spawn=original;syncBuiltinESMExports();}
 console.log('FINAL_ADMISSION_ARCHIVE',JSON.stringify({reads,archived,spawned,logs:launchLog(f)}));
 assert.equal(archived,true);assert.equal(spawned,0);
 assert.equal(h.runtime.threadsState('acme').find(r => r.thread === thread).holder, null);
 const db = new DatabaseSync(join(f.dataRoot, 'bus/bus.sqlite'), { readOnly: true });
 try { assert.equal(db.prepare("SELECT count(*) AS n FROM thread_claims WHERE thread=? AND op='claim'").get(thread).n, 0); } finally { db.close(); }
});