jason.woltjeandClaude Opus 5.5
4da6815293
Commits move as v2 bundles over bus-host/git.sock (decision 135). The bus
host keeps one bare working repository per business and project, seeds main
from the project root under a per-repository lock, fetches a pushed bundle
with fsck into a temp ref, requires history shared with main, and moves the
task's work ref only as a fast-forward. Host and runner git run with no
system or global config, hooks at /dev/null and local transports only. The
socket binds only for roles holding git.push.working or review.verdict.
The broker records a push as the action.allowed for git.push.working with
its commit, takes review.request for the latest push and one review.verdict
per request, and refuses task.close on a pushed task without an approve
verdict for the latest push. recordEvent reserves both shapes, so agents
can't forge either.
Reviewed by Dewey (28196) and Sage (28197), both approve. Gate on 7a729aef:
14 node suites and 9 scripts green.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
638 lines
34 KiB
JavaScript
638 lines
34 KiB
JavaScript
// The S6 launcher inside a real bus host: real broker child, real runners
|
|
// in the bwrap sandbox, the fake adapter from packages/harness standing in for the
|
|
// harness. The PM launches a coder through its `launch` tool; every host
|
|
// refusal comes back over the launch socket and lands in the launch log.
|
|
|
|
import { test } from "node:test";
|
|
import assert from "node:assert/strict";
|
|
import { spawn, spawnSync } from "node:child_process";
|
|
import { once } from "node:events";
|
|
import { createInterface } from "node:readline";
|
|
import { connect } from "node:net";
|
|
import { chmodSync, existsSync, mkdirSync, readFileSync, readdirSync, renameSync, rmSync, statSync, writeFileSync } from "node:fs";
|
|
import { join } from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import { Client } from "../../bus/src/client.mjs";
|
|
import { gitSocketPath, launchLogFile, launchSocketPath, readSessions, Registry, SESSION_CREDENTIALS, sessionsFile, workspaceDir } from "../../seat/src/session.mjs";
|
|
import { sandboxProblem } from "../../seat/src/sandbox.mjs";
|
|
import { gitEnv } from "../../seat/src/checkout.mjs";
|
|
import { bootConfig, loadSystem } from "../src/config.mjs";
|
|
import { startHost, startTimeOf } from "../src/host.mjs";
|
|
import { claudeVersion, createLauncher } from "../src/launcher.mjs";
|
|
import { mockAnthropic, REPO } from "../../harness/tests/helpers.mjs";
|
|
import { fixture, tmp } from "./helpers.mjs";
|
|
|
|
const FAKE = fileURLToPath(new URL("../../harness/tests/fixtures/fake-adapter.mjs", import.meta.url));
|
|
const HOST = fileURLToPath(new URL("./fixtures/launch-host.mjs", import.meta.url));
|
|
const skip = sandboxProblem() ?? false;
|
|
|
|
// pm opus, coder and reviewer sonnet, one of each family; cto is listed but
|
|
// runs the system model, which is in no family.
|
|
function edit(doc) {
|
|
doc.roles.coder.vars.model = "claude-sonnet-5-5";
|
|
doc.roles.reviewer.vars = { harness: "pi", model: "claude-sonnet-5-5" };
|
|
doc.launch = { by: "pm", instances: ["coder", "reviewer", "cto"], max: { opus: 1, sonnet: 1 } };
|
|
return doc;
|
|
}
|
|
|
|
async function until(fn, ms = 15000) {
|
|
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");
|
|
}
|
|
|
|
// git in a test repository, with nothing of the shell's git config.
|
|
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);
|
|
return r.stdout.trim();
|
|
}
|
|
function commit(dir, text) {
|
|
writeFileSync(join(dir, "file.txt"), `${text}\n`);
|
|
gitIn(dir, "add", "file.txt");
|
|
gitIn(dir, "commit", "-q", "-m", text);
|
|
return gitIn(dir, "rev-parse", "HEAD");
|
|
}
|
|
|
|
// A config and data root with the fake adapter; setup() starts a host on it.
|
|
// The session credential is a stand-in, and the fake adapter never uses it.
|
|
// Every credential variable of the shell running the tests is dropped.
|
|
function prepare(t, more = (doc) => doc) {
|
|
const root = tmp(t, "mosaic-launch-");
|
|
const f = fixture(root, "acme", (doc) => more(edit(doc)));
|
|
for (const k of SESSION_CREDENTIALS.flat()) delete f.env[k];
|
|
f.env.CLAUDE_CODE_OAUTH_TOKEN = "stand-in";
|
|
mkdirSync(f.dataRoot, { mode: 0o700 });
|
|
// The project root is a repository on main, so each launch can clone it.
|
|
const project = f.doc.projects.stack.root;
|
|
gitIn(project, "init", "-q", "-b", "main");
|
|
commit(project, "one");
|
|
const adapter = join(root, "adapter.sh");
|
|
writeFileSync(adapter, `exec '${process.execPath}' '${FAKE}'\n`, { mode: 0o644 });
|
|
return { root, f, adapter };
|
|
}
|
|
|
|
async function setup(t, more = (doc) => doc, { root, f, adapter } = prepare(t, more), options = {}) {
|
|
const system = loadSystem({ env: f.env });
|
|
const boot = bootConfig({ system, businessId: "acme", env: f.env });
|
|
const logs = [];
|
|
const host = await startHost({ boot, business: "acme", log: (l) => logs.push(l) });
|
|
let launcher;
|
|
try {
|
|
launcher = createLauncher({
|
|
host,
|
|
system,
|
|
env: f.env,
|
|
adapters: { pi: adapter, "claude-code": adapter },
|
|
sessionDefaults: { pollInterval: 100 },
|
|
versions: { pi: "0.85.1", claude: null },
|
|
readOnly: [FAKE],
|
|
log: (l) => logs.push(l),
|
|
...options,
|
|
});
|
|
await launcher.listen();
|
|
} catch (e) {
|
|
await host.close(1);
|
|
throw e;
|
|
}
|
|
let closed = false;
|
|
const close = async () => {
|
|
if (closed) return host.done;
|
|
closed = true;
|
|
return host.close(0);
|
|
};
|
|
t.after(close);
|
|
// A test caller: a cto launch bound to this process, claimed.
|
|
const cto = await host.bindLaunch({ business: "acme", role: "cto", run: "cto-test", harness: "claude-code", pid: process.pid, startTime: startTimeOf(process.pid) });
|
|
const client = new Client({ path: host.path, cap: cto.cap });
|
|
await client.call("role.claim");
|
|
return { root, f, host, launcher, cto, client, logs, close };
|
|
}
|
|
|
|
// One request on the launch socket, one reply line.
|
|
function ask(path, line) {
|
|
return new Promise((resolve, reject) => {
|
|
const s = connect(path);
|
|
let buf = "";
|
|
s.on("connect", () => s.write(line));
|
|
s.on("data", (b) => (buf += b));
|
|
s.on("end", () => resolve(JSON.parse(buf)));
|
|
s.on("error", reject);
|
|
});
|
|
}
|
|
|
|
const log = (f) => readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l));
|
|
|
|
test("bus start --pm: the PM runs under its own PID namespace, registered, with a manifest", { skip, timeout: 30000 }, async (t) => {
|
|
const { f, launcher, host, cto, close } = await setup(t);
|
|
const sock = launchSocketPath(f.dataRoot);
|
|
assert.equal(statSync(sock).mode & 0o777, 0o600);
|
|
|
|
const pm = await launcher.launchPm();
|
|
assert.deepEqual({ ...pm, run: undefined }, { run: undefined, instance: "pm", harness: "pi", model: "claude-opus-5-5" });
|
|
assert.match(pm.run, /^r[0-9a-f]{12}$/);
|
|
|
|
const [s] = readSessions(f.dataRoot);
|
|
assert.equal(s.run, pm.run);
|
|
assert.equal(s.live, true);
|
|
assert.equal(s.family, "opus");
|
|
assert.equal(s.launchedBy, "jason");
|
|
assert.notEqual(s.pid, s.runnerPid, "the runner is the bwrap process's child");
|
|
assert.equal(statSync(sessionsFile(f.dataRoot)).mode & 0o777, 0o600);
|
|
|
|
const runDir = s.runDir;
|
|
assert.equal(statSync(runDir).mode & 0o777, 0o700);
|
|
const session = JSON.parse(readFileSync(join(runDir, "session.json"), "utf8"));
|
|
assert.equal(session.launchedBy, "jason");
|
|
assert.equal(session.brokerSocket, host.path);
|
|
assert.equal(session.launchSocket, sock);
|
|
// Row 93: the PM neither pushes nor reviews, so no git socket; the launcher serves one.
|
|
assert.equal(session.gitSocket, null);
|
|
assert.equal(statSync(gitSocketPath(f.dataRoot)).mode & 0o777, 0o600);
|
|
const manifest = JSON.parse(readFileSync(join(runDir, "bundle", "manifest.json"), "utf8"));
|
|
assert.equal(manifest.piVersion, "0.85.1");
|
|
assert.equal(manifest.claudeVersion, null);
|
|
const [launched] = log(f).filter((e) => e.event === "launch");
|
|
assert.equal(launched.run, pm.run);
|
|
assert.equal(launched.manifest, join(runDir, "bundle", "manifest.json"));
|
|
assert.equal(manifest.model, "claude-opus-5-5");
|
|
|
|
// A second PM is refused before anything spawns.
|
|
await assert.rejects(launcher.launchPm(), /PM launch refused: instance-running/);
|
|
|
|
// The session's environment carries the run id and no capability; the
|
|
// capability is nowhere on disk or on any session process's command line.
|
|
await until(() => existsSync(join(runDir, "runner.log")) && readSessions(f.dataRoot)[0]?.live);
|
|
const proc = (pid, what) => {
|
|
try {
|
|
return readFileSync(`/proc/${pid}/${what}`, "utf8");
|
|
} catch {
|
|
return "";
|
|
}
|
|
};
|
|
const environ = proc(s.runnerPid, "environ").split("\0");
|
|
assert.ok(environ.includes(`MOSAIC_RUN_ID=${pm.run}`));
|
|
assert.ok(!environ.some((v) => v.startsWith("MOSAIC_CONFIG=")), "only allowlisted variables pass");
|
|
const files = (dir) => readdirSync(dir, { recursive: true, withFileTypes: true }).filter((d) => d.isFile()).map((d) => join(d.parentPath, d.name));
|
|
const capOf = cto.cap; // a known capability shape, to prove the scan finds nothing like it
|
|
for (const file of files(f.dataRoot)) {
|
|
const text = readFileSync(file, "latin1");
|
|
assert.ok(!text.includes(capOf), `${file} holds no capability`);
|
|
}
|
|
for (const pid of [s.pid, s.runnerPid]) {
|
|
assert.notEqual(proc(pid, "cmdline"), "");
|
|
assert.ok(!/[0-9a-f]{64}/.test(proc(pid, "cmdline")), "no capability in argv");
|
|
assert.ok(!/[0-9a-f]{64}/.test(proc(pid, "environ")), "no capability in the environment");
|
|
}
|
|
|
|
// Closing the host stops the session and ends the launch.
|
|
assert.equal(await close(), 0);
|
|
assert.equal(startTimeOf(s.runnerPid), null);
|
|
assert.deepEqual(readSessions(f.dataRoot), []);
|
|
assert.equal(existsSync(sock), false);
|
|
const end = log(f).find((e) => e.event === "end");
|
|
assert.equal(end.run, pm.run);
|
|
assert.equal(end.reason, "stopped");
|
|
assert.equal(end.exitCode, 0);
|
|
});
|
|
|
|
test("the PM launches a coder through its launch tool; the coder answers; refusals name their code", { skip, timeout: 30000 }, async (t) => {
|
|
const { f, launcher, client, cto } = await setup(t);
|
|
const sock = launchSocketPath(f.dataRoot);
|
|
const pm = await launcher.launchPm();
|
|
|
|
// cto asks pm to launch a coder; pm's turn calls the launch tool.
|
|
const sent = await client.call("message.send", { to: "pm", class: "REQUEST", body: 'TOOL launch {"instance":"coder"}' });
|
|
const reply = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id));
|
|
assert.equal(reply.from_role, "pm");
|
|
assert.equal(reply.class, "RESULT");
|
|
const result = JSON.parse(reply.body);
|
|
assert.equal(result.instance, "coder");
|
|
assert.equal(result.model, "claude-sonnet-5-5");
|
|
const coder = readSessions(f.dataRoot).find((s) => s.instance === "coder");
|
|
assert.equal(coder.run, result.run);
|
|
assert.equal(coder.launchedBy, "pm");
|
|
assert.equal(coder.live, true);
|
|
// Only the role that may launch has the launch socket in its sandbox.
|
|
const bound = (pid) => readFileSync(`/proc/${pid}/cmdline`, "utf8").split("\0").includes(sock);
|
|
assert.equal(bound(readSessions(f.dataRoot).find((s) => s.instance === "pm").pid), true);
|
|
assert.equal(bound(coder.pid), false);
|
|
// Row 93: and only a role that pushes or reviews has the git socket.
|
|
const gitBound = (pid) => readFileSync(`/proc/${pid}/cmdline`, "utf8").split("\0").includes(gitSocketPath(f.dataRoot));
|
|
assert.equal(gitBound(coder.pid), true);
|
|
assert.equal(gitBound(readSessions(f.dataRoot).find((s) => s.instance === "pm").pid), false);
|
|
assert.equal(JSON.parse(readFileSync(join(coder.runDir, "session.json"), "utf8")).gitSocket, gitSocketPath(f.dataRoot));
|
|
|
|
// The coder claimed its role and answers its own messages.
|
|
const hello = await client.call("message.send", { to: "coder", class: "REQUEST", body: "hello" });
|
|
const echo = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === hello.id));
|
|
assert.equal(echo.from_role, "coder");
|
|
assert.match(echo.body, /^echo: Message .* from cto, class REQUEST:\n\nhello/);
|
|
|
|
// Host refusals, asked with the cto capability (steps 2 to 8 come before
|
|
// the broker's role.launch check, which cto would fail).
|
|
const refusals = [
|
|
["pm", "instance-not-listed"],
|
|
["nobody", "instance-not-listed"],
|
|
["coder", "instance-running"],
|
|
["reviewer", "capacity-full"],
|
|
["cto", "unknown-model-family"],
|
|
];
|
|
for (const [instance, code] of refusals) {
|
|
assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance })}\n`), { ok: false, error: code }, instance);
|
|
}
|
|
// A listed, free instance with room still needs role.launch: cto lacks it.
|
|
// Stop the coder first so sonnet has room again.
|
|
const coderRun = coder.run;
|
|
process.kill(coder.runnerPid, "SIGTERM");
|
|
await until(() => !readSessions(f.dataRoot).some((s) => s.run === coderRun) && log(f).some((e) => e.event === "end" && e.run === coderRun));
|
|
const denied = await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: "reviewer" })}\n`);
|
|
assert.equal(denied.ok, false);
|
|
assert.notEqual(denied.error, "capacity-full");
|
|
|
|
// Malformed requests.
|
|
assert.deepEqual(await ask(sock, "not json\n"), { ok: false, error: "invalid-request" });
|
|
assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: "short", instance: "coder" })}\n`), { ok: false, error: "unauthenticated" });
|
|
assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: "0".repeat(64), instance: "coder" })}\n`), { ok: false, error: "unauthenticated" });
|
|
assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: 7 })}\n`), { ok: false, error: "invalid-request" });
|
|
|
|
// The launch log has the launches, every refusal, and the coder's end.
|
|
const entries = log(f);
|
|
assert.deepEqual(entries.filter((e) => e.event === "launch").map((e) => [e.instance, e.launchedBy]), [["pm", "jason"], ["coder", "pm"]]);
|
|
const refused = entries.filter((e) => e.event === "refused").map((e) => [e.instance, e.code]);
|
|
for (const r of refusals) assert.ok(refused.some(([i, c]) => i === r[0] && c === r[1]), `${r} is logged`);
|
|
const end = entries.find((e) => e.event === "end" && e.run === coderRun);
|
|
assert.equal(end.reason, "stopped");
|
|
assert.ok(readSessions(f.dataRoot).some((s) => s.run === pm.run));
|
|
});
|
|
|
|
// Row 91 (#1585): the project clone in each workspace.
|
|
test("each launch makes sure the workspace has a clone of the project: cloned, fast-forwarded, or kept and logged", { skip, timeout: 60000 }, async (t) => {
|
|
const given = prepare(t);
|
|
const project = given.f.doc.projects.stack.root;
|
|
const clone = join(workspaceDir(given.f.dataRoot, "acme", "pm"), "stack");
|
|
const prompt = (run) => readFileSync(join(given.f.dataRoot, "launches", "acme", run, "bundle", "prompt.md"), "utf8");
|
|
const kept = () => log(given.f).filter((e) => e.event === "checkout-kept");
|
|
|
|
// The first launch clones the root's branch, with the root as origin.
|
|
const first = await setup(t, undefined, given);
|
|
const one = gitIn(project, "rev-parse", "HEAD");
|
|
const pm1 = await first.launcher.launchPm();
|
|
assert.equal(gitIn(clone, "rev-parse", "HEAD"), one);
|
|
assert.equal(gitIn(clone, "symbolic-ref", "--short", "HEAD"), "main");
|
|
assert.equal(gitIn(clone, "config", "--get-all", "remote.origin.url"), project);
|
|
assert.ok(prompt(pm1.run).includes(`- Project clones: stack at ${clone}, on branch main. `));
|
|
|
|
// While the PM runs, a second launch is refused before the host runs git
|
|
// in its workspace: the root moved on, and the clone didn't.
|
|
const two = commit(project, "two");
|
|
await assert.rejects(first.launcher.launchPm(), /PM launch refused: instance-running/);
|
|
assert.equal(gitIn(clone, "rev-parse", "HEAD"), one);
|
|
await first.client.call("role.release"); // the test's own cto claim, so the next host's can claim again
|
|
assert.equal(await first.close(), 0);
|
|
|
|
// The next launch fast-forwards the clean clone.
|
|
const second = await setup(t, undefined, given);
|
|
const pm2 = await second.launcher.launchPm();
|
|
assert.equal(gitIn(clone, "rev-parse", "HEAD"), two);
|
|
assert.ok(prompt(pm2.run).includes(`- Project clones: stack at ${clone}, on branch main. `));
|
|
assert.deepEqual(kept(), []);
|
|
await second.client.call("role.release");
|
|
assert.equal(await second.close(), 0);
|
|
|
|
// A clone with the session's own work in it is kept, and the log says why.
|
|
writeFileSync(join(clone, "notes.txt"), "mine\n");
|
|
const three = commit(project, "three");
|
|
const third = await setup(t, undefined, given);
|
|
const pm3 = await third.launcher.launchPm();
|
|
assert.equal(gitIn(clone, "rev-parse", "HEAD"), two);
|
|
assert.equal(readFileSync(join(clone, "notes.txt"), "utf8"), "mine\n");
|
|
assert.deepEqual(kept().map(({ instance, project: p, reason }) => ({ instance, project: p, reason })), [{ instance: "pm", project: "stack", reason: "dirty" }]);
|
|
assert.ok(prompt(pm3.run).includes(`- Project clones: stack at ${clone}, not updated at this launch (dirty). `));
|
|
assert.equal(await third.close(), 0);
|
|
|
|
// The root was never written: its branch, head and tree are the test's.
|
|
assert.equal(gitIn(project, "rev-parse", "HEAD"), three);
|
|
assert.equal(gitIn(project, "for-each-ref", "--format=%(refname)"), "refs/heads/main");
|
|
assert.equal(gitIn(project, "status", "--porcelain"), "");
|
|
});
|
|
|
|
test("a project clone that can't be made refuses checkout-failed, from --pm and from a session, with no run and no clone left", { skip, timeout: 30000 }, async (t) => {
|
|
const given = prepare(t);
|
|
renameSync(join(given.f.doc.projects.stack.root, ".git"), join(given.root, "away.git"));
|
|
const { f, launcher, host } = await setup(t, undefined, given);
|
|
await assert.rejects(launcher.launchPm(), (e) => e.exitCode === 3 && /^PM launch refused: checkout-failed \(stack: the root .* can't be read: /.test(e.message));
|
|
|
|
// A pm capability bound to this process asks for the coder.
|
|
const pm = await host.bindLaunch({ business: "acme", role: "pm", run: "pm-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
|
|
await new Client({ path: host.path, cap: pm.cap }).call("role.claim");
|
|
assert.deepEqual(await launcher.fromSession(pm.cap, "coder"), { ok: false, error: "checkout-failed" });
|
|
|
|
const entries = log(f);
|
|
assert.deepEqual(entries.map((e) => [e.event, e.instance, e.code]), [["refused", "pm", "checkout-failed"], ["refused", "coder", "checkout-failed"]]);
|
|
for (const e of entries) assert.match(e.why, /^stack: the root /);
|
|
assert.deepEqual(readSessions(f.dataRoot), []);
|
|
assert.deepEqual(readdirSync(join(f.dataRoot, "launches")), ["acme.jsonl"], "no run was made");
|
|
for (const instance of ["pm", "coder"]) assert.deepEqual(readdirSync(workspaceDir(f.dataRoot, "acme", instance)), [], `${instance}: no clone, no temp`);
|
|
});
|
|
|
|
test("a runner that stops at once ends its launch with the runner's reason", { skip, timeout: 30000 }, async (t) => {
|
|
// pm's tracker token has expired: the runner's founder-credential stop
|
|
// exits 20 before role.claim, and the host records founder-credentials.
|
|
const { f, launcher } = await setup(t, (doc) => {
|
|
doc.roles.pm.credentials.vikunja.expires = "2000-01-01";
|
|
return doc;
|
|
});
|
|
const pm = await launcher.launchPm();
|
|
const end = await until(() => existsSync(launchLogFile(f.dataRoot, "acme")) && log(f).find((e) => e.event === "end" && e.run === pm.run));
|
|
assert.equal(end.reason, "founder-credentials");
|
|
assert.equal(end.exitCode, 20);
|
|
assert.deepEqual(readSessions(f.dataRoot), []);
|
|
assert.match(readFileSync(join(f.dataRoot, "launches", "acme", pm.run, "runner.log"), "utf8"), /vikunja/);
|
|
});
|
|
|
|
test("a Claude Code launch with no session credential refuses before it starts", { timeout: 30000 }, async (t) => {
|
|
const given = prepare(t, (doc) => {
|
|
doc.roles.pm.vars.harness = "claude-code";
|
|
return doc;
|
|
});
|
|
delete given.f.env.CLAUDE_CODE_OAUTH_TOKEN;
|
|
given.f.env.ANTHROPIC_AUTH_TOKEN = "stand-in"; // a proxy key with no proxy
|
|
const { f, launcher } = await setup(t, undefined, given);
|
|
const names = "CLAUDE_CODE_OAUTH_TOKEN, ANTHROPIC_API_KEY, ANTHROPIC_AUTH_TOKEN with ANTHROPIC_BASE_URL";
|
|
await assert.rejects(launcher.launchPm(), (e) => e.exitCode === 3 && e.message === `PM launch refused: no-session-credential (the bus host has none of ${names}; the bus unit's EnvironmentFile sets one when the bus starts)`);
|
|
assert.deepEqual(log(f).map((e) => [e.event, e.code]), [["refused", "no-session-credential"]]);
|
|
assert.deepEqual(readSessions(f.dataRoot), []);
|
|
assert.deepEqual(readdirSync(join(f.dataRoot, "launches")), ["acme.jsonl"], "no run was made");
|
|
});
|
|
|
|
test("a pi launch needs no session credential: pi logs in through its own ENV_ALLOW variables", { skip, timeout: 30000 }, async (t) => {
|
|
const given = prepare(t);
|
|
delete given.f.env.CLAUDE_CODE_OAUTH_TOKEN;
|
|
assert.deepEqual(SESSION_CREDENTIALS.flat().filter((k) => k in given.f.env), []);
|
|
const { f, launcher } = await setup(t, undefined, given);
|
|
const pm = await launcher.launchPm();
|
|
assert.equal(pm.harness, "pi");
|
|
assert.deepEqual(log(f).map((e) => [e.event, e.instance]), [["launch", "pm"]]);
|
|
});
|
|
|
|
// The runner is pid 1 in its namespace and ignores a SIGKILL sent from
|
|
// inside it, so the host kills it by its host pid. bwrap then exits 137,
|
|
// which the launcher reads back as the signal.
|
|
test("a runner killed from the host ends its launch as killed, not failed", { skip, timeout: 30000 }, async (t) => {
|
|
const { f, launcher } = await setup(t);
|
|
const pm = await launcher.launchPm();
|
|
const runnerLog = join(f.dataRoot, "launches", "acme", pm.run, "runner.log");
|
|
await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
|
|
const [s] = readSessions(f.dataRoot);
|
|
process.kill(s.runnerPid, "SIGKILL");
|
|
const end = await until(() => log(f).find((e) => e.event === "end" && e.run === pm.run));
|
|
assert.deepEqual([end.reason, end.exitCode, end.signal], ["killed", null, "SIGKILL"]);
|
|
assert.deepEqual(readSessions(f.dataRoot), []);
|
|
});
|
|
|
|
test("a missing or failing bwrap refuses sandbox-unavailable before anything starts; there is no unconfined fallback",{ timeout: 30000 }, async (t) => {
|
|
const given = prepare(t);
|
|
const bad = join(given.root, "bwrap");
|
|
writeFileSync(bad, "#!/bin/sh\necho 'bwrap: setting up uid map: Permission denied' >&2\nexit 1\n", { mode: 0o755 });
|
|
const { f, launcher, cto } = await setup(t, undefined, given, { bwrap: bad });
|
|
await assert.rejects(launcher.launchPm(), (e) => e.message === `PM launch refused: sandbox-unavailable (${bad} failed: bwrap: setting up uid map: Permission denied)`);
|
|
rmSync(bad);
|
|
await assert.rejects(launcher.launchPm(), (e) => e.message === `PM launch refused: sandbox-unavailable (${bad} is missing)`);
|
|
// A session's launch request gets the same code, through refuse().
|
|
assert.deepEqual(await launcher.fromSession(cto.cap, "coder"), { ok: false, error: "sandbox-unavailable" });
|
|
assert.deepEqual(log(f).map((e) => e.code), ["sandbox-unavailable", "sandbox-unavailable", "sandbox-unavailable"]);
|
|
assert.deepEqual(readSessions(f.dataRoot), []);
|
|
assert.deepEqual(readdirSync(join(f.dataRoot, "launches")), ["acme.jsonl"], "no run was made");
|
|
});
|
|
|
|
// The brief's one real Claude Code launch: the host's claude in a coder's
|
|
// sandbox, logged in with a stand-in proxy key against the scripted Messages
|
|
// API, runs one Bash call that tries the founder's stand-ins, another run
|
|
// and the checkout. No real credential and no real model.
|
|
const which = spawnSync("sh", ["-c", "command -v claude"], { encoding: "utf8" });
|
|
const CLAUDE = which.status === 0 ? which.stdout.trim() : null;
|
|
|
|
test("a real Claude Code coder's shell can't read a stand-in secret or anything else its sandbox doesn't hold", { skip: skip || (CLAUDE ? false : "claude is not on PATH"), timeout: 180000 }, async (t) => {
|
|
const script = [];
|
|
const api = await mockAnthropic(t, { script, done: (r) => `ANSWER ${JSON.stringify(r.map((x) => [x.isError, x.content.slice(0, 1500)]))}` });
|
|
const given = prepare(t, (doc) => {
|
|
doc.roles.coder.vars.harness = "claude-code";
|
|
doc.roles.coder.vars["limits.tools"] = ["bash"];
|
|
return doc;
|
|
});
|
|
delete given.f.env.CLAUDE_CODE_OAUTH_TOKEN;
|
|
Object.assign(given.f.env, { ANTHROPIC_AUTH_TOKEN: "stand-in", ANTHROPIC_BASE_URL: api.url, MOSAIC_CLAUDE_BIN: CLAUDE });
|
|
const secret = join(given.root, "home", ".config", "mosaic-dev", "secrets", "stand-in.env");
|
|
const mosaic = join(given.root, "home", ".mosaic", "fleet", "token");
|
|
for (const p of [secret, mosaic]) {
|
|
mkdirSync(join(p, ".."), { recursive: true, mode: 0o700 });
|
|
writeFileSync(p, "FAKE_TOKEN=stand-in-value\n", { mode: 0o600 });
|
|
}
|
|
const claudeAdapter = join(REPO, "adapters", "claude", "adapter.sh");
|
|
const { f, launcher, client } = await setup(t, undefined, given, { adapters: { pi: given.adapter, "claude-code": claudeAdapter } });
|
|
const pm = await launcher.launchPm();
|
|
const pmRun = readSessions(f.dataRoot).find((s) => s.run === pm.run).runDir;
|
|
|
|
// [what, shell command, may the session do it]
|
|
const probes = [
|
|
["read the stand-in secret", `cat '${secret}'`, false],
|
|
["read the stand-in ~/.mosaic", `cat '${mosaic}'`, false],
|
|
["read the PM's run directory", `cat '${pmRun}/session.json'`, false],
|
|
["read the data root's sessions file", `cat '${sessionsFile(f.dataRoot)}'`, false],
|
|
["read the checkout", `cat '${join(REPO, "AGENTS.md")}'`, false],
|
|
["write outside", `touch '${given.root}/escaped'`, false],
|
|
["write and read its workspace", "echo ws > probe && cat probe", true],
|
|
["write and read /tmp", "echo tmp > /tmp/probe && cat /tmp/probe", true],
|
|
["HOME is the run's home/", `case "$HOME" in */home) test -w "$HOME";; *) false;; esac`, true],
|
|
];
|
|
const command = probes.map(([what, cmd], i) => `if { ${cmd}; } >/dev/null 2>&1; then echo "${i} yes"; else echo "${i} no"; fi`).join("\n");
|
|
script.push({ name: "Bash", input: { command, description: "probe the sandbox" } });
|
|
|
|
const sent = await client.call("message.send", { to: "pm", class: "REQUEST", body: 'TOOL launch {"instance":"coder"}' });
|
|
const launched = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id));
|
|
assert.equal(JSON.parse(launched.body).instance, "coder", launched.body);
|
|
const coder = readSessions(f.dataRoot).find((s) => s.instance === "coder");
|
|
assert.equal(coder.harness, "claude-code");
|
|
|
|
const hello = await client.call("message.send", { to: "coder", class: "REQUEST", body: "probe" });
|
|
const reply = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === hello.id), 150000);
|
|
assert.equal(reply.from_role, "coder");
|
|
const answer = reply.body.match(/ANSWER (\[.*\])/s);
|
|
assert.ok(answer, reply.body);
|
|
const [[isError, out]] = JSON.parse(answer[1]);
|
|
assert.equal(isError, false, out);
|
|
const got = Object.fromEntries(out.trim().split("\n").map((l) => l.split(" ")).map(([i, v]) => [probes[i][0], v === "yes"]));
|
|
assert.deepEqual(got, Object.fromEntries(probes.map(([what, , may]) => [what, may])));
|
|
assert.ok(!existsSync(join(given.root, "escaped")));
|
|
assert.ok(!reply.body.includes("stand-in-value"), "the secret's value never comes back");
|
|
// The session logged in with the stand-in proxy key, against the mock.
|
|
assert.ok(api.requests.length > 0);
|
|
assert.ok(api.requests.some((r) => r.tools.includes("Bash")), api.requests.map((r) => r.tools.join(",")).join(" | "));
|
|
});
|
|
|
|
for (const exited of [false, true]) {
|
|
test(`a host that died hard leaves its sessions to the next host, which ${exited ? "finds one already exited (23)" : "kills them"} and ends their runs so the role can be launched again`, { skip, timeout: 30000 }, async (t) => {
|
|
const given = prepare(t);
|
|
const { f } = given;
|
|
const old = spawn(process.execPath, [HOST, given.adapter, FAKE], { env: f.env, stdio: ["ignore", "pipe", "inherit"] });
|
|
t.after(() => old.kill("SIGKILL"));
|
|
const [line] = await once(createInterface({ input: old.stdout }), "line");
|
|
const first = JSON.parse(line);
|
|
const [left] = readSessions(f.dataRoot);
|
|
assert.equal(left.run, first.run);
|
|
// Killed before the claim, the runner would exit 23 on its own.
|
|
const runnerLog = join(f.dataRoot, "launches", "acme", first.run, "runner.log");
|
|
await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
|
|
assert.equal(left.live, true);
|
|
old.kill("SIGKILL");
|
|
await once(old, "exit");
|
|
// The broker child exits on the lost IPC channel; the session runs on.
|
|
await until(() => !existsSync(join(f.dataRoot, "bus", "writer.lock")));
|
|
assert.equal(startTimeOf(left.runnerPid), left.runnerStartTime);
|
|
if (!exited) {
|
|
// Stopped, the leftover neither polls the dead broker nor answers
|
|
// SIGTERM: only the next host's SIGKILL ends it.
|
|
for (const pid of [left.pid, left.runnerPid]) process.kill(pid, "SIGSTOP");
|
|
t.after(() => {
|
|
for (const [pid, start] of [[left.pid, left.startTime], [left.runnerPid, left.runnerStartTime]]) {
|
|
if (startTimeOf(pid) === start) process.kill(pid, "SIGKILL");
|
|
}
|
|
});
|
|
}
|
|
if (exited) {
|
|
await until(() => startTimeOf(left.pid) === null);
|
|
assert.match(readFileSync(runnerLog, "utf8"), /broker unreachable after 30 tries[\s\S]*exiting 23/);
|
|
assert.equal(readSessions(f.dataRoot)[0].live, false);
|
|
}
|
|
|
|
const s = await setup(t, undefined, given);
|
|
assert.equal(startTimeOf(left.pid), null, "the old session process is gone");
|
|
assert.equal(startTimeOf(left.runnerPid), null, "and its runner with it");
|
|
assert.ok(s.logs.includes(`run ${first.run} (pm) from an earlier host ended: host-lost`), s.logs.join("\n"));
|
|
const ended = (await s.client.call("trail", { subject: first.run })).filter((r) => r.kind === "session.ended");
|
|
assert.equal(ended.length, 1);
|
|
assert.deepEqual(ended[0].body, { reason: "host-lost", exitCode: null, released: true });
|
|
const log = readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l));
|
|
assert.deepEqual(log.filter((e) => e.event === "end").map((e) => [e.run, e.reason]), [[first.run, "host-lost"]]);
|
|
assert.deepEqual(readSessions(f.dataRoot), []);
|
|
|
|
// Without the end, the old run would still hold pm and this runner would exit 21.
|
|
const again = await s.launcher.launchPm();
|
|
await until(async () => (await s.client.call("agents")).some((a) => a.role === "pm" && a.holder_run === again.run));
|
|
await s.client.call("role.release"); // the test's own cto claim, so the next host's can claim again
|
|
assert.equal(await s.close(), 0);
|
|
|
|
// Died after the end but before sessions.json changed: the run stays ended.
|
|
const { live, ...entry } = left;
|
|
new Registry(f.dataRoot).add(entry);
|
|
const third = await setup(t, undefined, given);
|
|
assert.ok(third.logs.includes(`run ${first.run} (pm) from an earlier host was already ended`), third.logs.join("\n"));
|
|
assert.equal((await third.client.call("trail", { subject: first.run })).filter((r) => r.kind === "session.ended").length, 1);
|
|
assert.equal(await third.close(), 0);
|
|
});
|
|
}
|
|
|
|
test("a malformed sessions.json refuses the host start with exit 3 and stays as it was", { timeout: 30000 }, async (t) => {
|
|
const given = prepare(t);
|
|
const file = sessionsFile(given.f.dataRoot);
|
|
mkdirSync(join(given.f.dataRoot, "bus-host"), { mode: 0o700 });
|
|
writeFileSync(file, "{", { mode: 0o600 });
|
|
await assert.rejects(setup(t, undefined, given), (e) => e.exitCode === 3 && /sessions file is unreadable.*move the file aside/.test(e.message));
|
|
assert.equal(readFileSync(file, "utf8"), "{");
|
|
assert.equal(existsSync(join(given.f.dataRoot, "bus", "writer.lock")), false, "the host closed");
|
|
});
|
|
|
|
const within = (p, ms, what) => Promise.race([p, new Promise((_, reject) => setTimeout(() => reject(new Error(`${what} took over ${ms} ms`)), ms).unref())]);
|
|
|
|
test("an over-long launch request is refused at once, not at the 10 s idle timeout", { timeout: 30000 }, async (t) => {
|
|
const { f } = await setup(t);
|
|
const s = connect(launchSocketPath(f.dataRoot));
|
|
t.after(() => s.destroy());
|
|
let buf = "";
|
|
s.on("data", (b) => (buf += b));
|
|
s.write("x".repeat(5000));
|
|
await within(once(s, "end"), 3000, "the refusal");
|
|
assert.deepEqual(JSON.parse(buf), { ok: false, error: "invalid-request" });
|
|
});
|
|
|
|
test("a launch client that never closes its side doesn't hold the host's close", { timeout: 30000 }, async (t) => {
|
|
const { f, close } = await setup(t);
|
|
const path = launchSocketPath(f.dataRoot);
|
|
// One got its reply and never ends; one never sends anything.
|
|
const answered = connect({ path, allowHalfOpen: true });
|
|
const silent = connect({ path, allowHalfOpen: true });
|
|
const connected = once(silent, "connect");
|
|
const gone = Promise.all([once(answered, "close"), once(silent, "close")]);
|
|
const drop = () => {
|
|
answered.destroy();
|
|
silent.destroy();
|
|
};
|
|
t.after(drop);
|
|
let buf = "";
|
|
answered.on("data", (b) => (buf += b));
|
|
answered.write("not json\n");
|
|
await once(answered, "end");
|
|
assert.deepEqual(JSON.parse(buf), { ok: false, error: "invalid-request" });
|
|
await within(connected, 2000, "connect");
|
|
// On a failure the clients go, so the host can still close and the test
|
|
// fails instead of hanging.
|
|
const code = await within(close(), 5000, "close").catch((e) => {
|
|
drop();
|
|
throw e;
|
|
});
|
|
assert.equal(code, 0);
|
|
assert.equal(existsSync(path), false);
|
|
answered.end();
|
|
silent.end();
|
|
await within(gone, 2000, "the clients' close");
|
|
});
|
|
|
|
test("a runner that ignores SIGTERM is killed when the host closes", { skip, timeout: 60000 }, async (t) => {
|
|
const { f, launcher, close } = await setup(t, undefined, undefined, { stopTimeoutMs: 1000 });
|
|
const pm = await launcher.launchPm();
|
|
const runnerLog = join(f.dataRoot, "launches", "acme", pm.run, "runner.log");
|
|
await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
|
|
const [s] = readSessions(f.dataRoot);
|
|
// Stopped from outside its namespace, it can't act on SIGTERM.
|
|
process.kill(s.runnerPid, "SIGSTOP");
|
|
const kill = () => {
|
|
if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGKILL");
|
|
};
|
|
t.after(kill);
|
|
const started = Date.now();
|
|
const code = await within(close(), 15000, "close").catch((e) => {
|
|
kill();
|
|
throw e;
|
|
});
|
|
assert.equal(code, 0);
|
|
assert.ok(Date.now() - started >= 1000, "it waited for the stop timeout");
|
|
assert.equal(startTimeOf(s.pid), null);
|
|
assert.equal(startTimeOf(s.runnerPid), null);
|
|
assert.deepEqual(readSessions(f.dataRoot), []);
|
|
const ends = log(f).filter((e) => e.event === "end" && e.run === pm.run);
|
|
assert.equal(ends.length, 1);
|
|
assert.equal(ends[0].signal, "SIGKILL");
|
|
});
|
|
|
|
// Row 68: the manifest's claudeVersion asks the binary sessions run.
|
|
test("claudeVersion asks MOSAIC_CLAUDE_BIN when set, else claude on PATH", (t) => {
|
|
const dir = tmp(t);
|
|
mkdirSync(join(dir, "bin"));
|
|
const stand = (p, v) => {
|
|
writeFileSync(p, `#!/bin/sh\necho "${v} (Claude Code)"\n`);
|
|
chmodSync(p, 0o755);
|
|
return p;
|
|
};
|
|
stand(join(dir, "bin", "claude"), "1.0.0");
|
|
const pinned = stand(join(dir, "pinned"), "9.9.9");
|
|
const PATH = join(dir, "bin");
|
|
assert.equal(claudeVersion({ PATH, MOSAIC_CLAUDE_BIN: pinned }), "9.9.9 (Claude Code)");
|
|
assert.equal(claudeVersion({ PATH }), "1.0.0 (Claude Code)");
|
|
assert.equal(claudeVersion({ PATH: join(dir, "none") }), null);
|
|
});
|