jason.woltje
39eadf8ba8
The cli tests that launch pi deleted only the session login names, so pi's preflight passed or refused by whatever provider keys the host exported. A shared helper now deletes every credential pi's preflight reads and gives pi launches one made-up key. The refusal itself is unchanged. pi bundle() takes a default argument again, so a call with none returns the non-proxy bundle, as before P3. pi.bundle leaves out a launch.models entry whose provider has no proxy route instead of failing every pi launch (F2); malformed proxy entries still refuse. The invalid session proxy context BundleError gets the code invalid-proxy-context. New tests cover F2, the code, and the three proxy account branches in the launcher (F3).
399 lines
20 KiB
JavaScript
399 lines
20 KiB
JavaScript
// Clear context (row 271, 264d, #1776, T4): the launcher half of D7 and D8. A clear of a running
|
|
// session stops its drained run once, retires it on the run's own end, and starts the successor
|
|
// through the same check()/start() as any session; a held start waits for capacity; an earlier
|
|
// host's live run of the cleared session holds the clear until it exits; and a session runs in its
|
|
// own pid namespace (N6); a bus that can't answer holds a clear where it is (T6). Real runners in
|
|
// the bwrap sandbox, with clear-adapter.mjs as the turn.
|
|
// The bus half is packages/bus/tests/session-clear.test.mjs.
|
|
//
|
|
// The host is session-launch.test.mjs's, with the five clear ops the broker child forwards.
|
|
|
|
import { test } from "node:test";
|
|
import assert from "node:assert/strict";
|
|
import { spawn, 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 { gitEnv } from "../../seat/src/checkout.mjs";
|
|
import { startTimeOf } from "../../seat/src/proc.mjs";
|
|
import { BWRAP, 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/clear-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");
|
|
}
|
|
// A mutant's kill is an assertion failure, not a timeout (sage-2707): the waits return what they
|
|
// last saw and the assertions after them 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; sonnet is its model's limit.
|
|
function prepare(t, { sonnet = 3, idle = 30 } = {}) {
|
|
const root = tmp(t, "mosaic-clear-");
|
|
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 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 };
|
|
}
|
|
|
|
async function start(t, { f, adapter }, { tickMs = 100, stopTimeoutMs } = {}) {
|
|
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),
|
|
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),
|
|
sessionClearExited: (m) => runtime.sessionClearExited(m.record),
|
|
sessionClearUncertain: (m) => runtime.sessionClearUncertain(m.record),
|
|
sessionClearRetire: (m) => runtime.sessionClearRetire(m.record),
|
|
sessionClearStart: (m) => runtime.sessionClearStart(m.record),
|
|
};
|
|
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,
|
|
...(stopTimeoutMs ? { stopTimeoutMs } : {}),
|
|
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 events = (f, event) => launchLog(f).filter((e) => e.event === event);
|
|
const sessionWs = (f, session) => join(f.dataRoot, "session-workspaces", "acme", "assistant", session);
|
|
const historyOf = (f, thread, session) => {
|
|
const p = join(f.dataRoot, "threads", "acme", thread, "sessions", session, "session", "history");
|
|
return existsSync(p) ? readFileSync(p, "utf8") : null;
|
|
};
|
|
|
|
function send(call, thread, session, body) {
|
|
const sent = call("thread.send", { thread, ...(session ? { session } : {}), body });
|
|
const reply = async () => {
|
|
const r = await until(() => call("thread.messages", { thread, ...(session ? { session } : {}) }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT"));
|
|
assert.equal(r.from_role, "assistant");
|
|
return JSON.parse(r.body.match(/THREAD (\{.*\})/)[1]);
|
|
};
|
|
return { sent, reply };
|
|
}
|
|
|
|
function thread(call) {
|
|
const project = call("project.create", { name: "stack work", repo: "stack" }).id;
|
|
const A = call("thread.create", { project, name: "alpha" }).id;
|
|
const [{ id: D }] = call("session.list", { thread: A });
|
|
return { project, A, D };
|
|
}
|
|
|
|
test("REQ-THREAD-2 T4 D7 D8 N6: a clear of a running session stops its drained run once, retires it on the run's end, and starts the successor fresh in its own namespace", { skip, timeout: 90000 }, async (t) => {
|
|
const prepared = prepare(t);
|
|
const { f } = prepared;
|
|
const host = await start(t, prepared);
|
|
const { A, D } = thread(host.call);
|
|
const first = await send(host.call, A, D, "before the clear").reply();
|
|
assert.deepEqual([first.before, first.cwd], [[], sessionWs(f, D)]);
|
|
// N6: the run sees its own pid namespace. Pid 1 there is the sandboxed command, not bwrap, and
|
|
// the host's processes aren't listed.
|
|
assert.notEqual(first.pid1[0], BWRAP);
|
|
assert.ok(first.pids.length < 20, JSON.stringify(first.pids));
|
|
assert.equal(first.pids.includes(1), true);
|
|
const [{ run }] = events(f, "launch");
|
|
|
|
const op = host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" });
|
|
const S = op.successor;
|
|
const done = await settle(() => {
|
|
const e = host.call("session.operation", { thread: A, op: "clear-1" });
|
|
return e.start === "started" && e;
|
|
}, 30000);
|
|
assert.deepEqual([done.state, done.start, done.refusal], ["retired", "started", null]);
|
|
// The drained run got one stop, and ended stopped; the clear retired it after that end.
|
|
const stops = events(f, "session-clear-stop");
|
|
assert.deepEqual(stops.map((e) => [e.session, e.run, e.op]), [[D, run, "clear-1"]]);
|
|
assert.deepEqual(events(f, "session-clear-uncertain"), []);
|
|
const kinds = launchLog(f).map((e) => [e.event, e.session ?? null]).filter(([e]) => ["session-clear-stop", "end", "session-clear-retired", "launch"].includes(e));
|
|
assert.deepEqual(kinds, [
|
|
["launch", D],
|
|
["session-clear-stop", D],
|
|
["end", D],
|
|
["session-clear-retired", D],
|
|
["launch", S],
|
|
]);
|
|
assert.deepEqual(events(f, "session-clear-retired").map((e) => [e.successor, e.op]), [[S, "clear-1"]]);
|
|
// The successor is the default now; a thread-only send reaches it, with no history and its own
|
|
// workspace. The old session's store stays as it was.
|
|
const after = await send(host.call, A, null, "after the clear").reply();
|
|
assert.deepEqual([after.body, after.before, after.cwd], ["after the clear", [], sessionWs(f, S)]);
|
|
assert.equal(historyOf(f, A, D), "before the clear\n");
|
|
assert.equal(historyOf(f, A, S), "after the clear\n");
|
|
assert.deepEqual(host.call("session.list", { thread: A }).map((r) => [r.id, r.state, r.default]), [
|
|
[D, "retired", false],
|
|
[S, "active", true],
|
|
]);
|
|
assert.deepEqual(readSessions(f.dataRoot).map((s) => s.session), [S]);
|
|
});
|
|
|
|
test("REQ-THREAD-2 T4 D8: a successor start refused capacity-full is held, reported and logged once, and starts when capacity frees", { skip, timeout: 90000 }, async (t) => {
|
|
const prepared = prepare(t, { sonnet: 1, idle: 4 });
|
|
const { f } = prepared;
|
|
const host = await start(t, prepared);
|
|
const { A, D } = thread(host.call);
|
|
const N = host.call("session.create", { thread: A, name: "side", op: "op-side" }).id;
|
|
// The side session takes the one sonnet slot; the default never ran, so its clear has no run.
|
|
await send(host.call, A, N, "fill the slot").reply();
|
|
const S = host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" }).successor;
|
|
const held = await settle(() => {
|
|
const e = host.call("session.operation", { thread: A, op: "clear-1" });
|
|
return e.start === "held" && e;
|
|
});
|
|
assert.deepEqual([held.state, held.start, held.refusal], ["retired", "held", null]);
|
|
assert.deepEqual(events(f, "session-clear-stop"), [], "no run to stop");
|
|
// The side session idles out, and the held start goes on the next tick after.
|
|
const started = await settle(() => {
|
|
const e = host.call("session.operation", { thread: A, op: "clear-1" });
|
|
return e.start === "started" && e;
|
|
}, 30000);
|
|
assert.equal(started.start, "started");
|
|
assert.deepEqual(events(f, "launch").map((e) => e.session), [N, S]);
|
|
const lines = host.logs.filter((l) => l.startsWith("clear clear-1:"));
|
|
assert.deepEqual(lines, [`clear clear-1: successor ${S} (assistant): start held: capacity-full`]);
|
|
assert.deepEqual(events(f, "refused").map((e) => [e.session, e.code]), [[S, "capacity-full"]]);
|
|
});
|
|
|
|
// sage-2830, Q2: after a restart, an earlier host's run of the cleared session that outlived
|
|
// recover() was ended host-lost, but it still runs. Its exit, not that end, is the proof, and it
|
|
// isn't signalled. A sleep child stands in for its runner, as in session-launch.test.mjs's Q2 test.
|
|
test("REQ-THREAD-2 T4 D7 Q2: after a restart, a live survivor of the cleared run holds the clear at drained with no signal, and its exit lets it retire", { skip, timeout: 90000 }, async (t) => {
|
|
const prepared = prepare(t);
|
|
const { f } = prepared;
|
|
const one = await start(t, prepared);
|
|
const { A, D } = thread(one.call);
|
|
const pid = 2 ** 22 + 7;
|
|
const record = { business: "acme", role: "assistant", run: "run-left", harness: "pi", pid, startTime: "1", thread: A, session: D };
|
|
const left = one.runtime.bindLaunch(record);
|
|
one.runtime.broker.request(left.cap, { verb: "role.claim" });
|
|
one.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" });
|
|
assert.deepEqual(one.runtime.broker.request(left.cap, { verb: "message.receive" }), []);
|
|
assert.deepEqual(one.runtime.sessionClears("acme").map((c) => [c.stage, c.run]), [["drained", "run-left"]]);
|
|
await one.close();
|
|
|
|
const runner = spawn("sleep", ["60"], { stdio: "ignore" });
|
|
const gone = new Promise((r) => runner.once("exit", r));
|
|
t.after(() => runner.kill("SIGKILL"));
|
|
const runnerStartTime = await until(() => startTimeOf(runner.pid));
|
|
new Registry(f.dataRoot).add({ business: "acme", instance: "assistant", thread: A, session: D, run: "run-left", harness: "pi", pid, startTime: "1", runnerPid: runner.pid, runnerStartTime });
|
|
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"));
|
|
// Ended in the bus, live on the host: the clear waits, and nothing is signalled.
|
|
for (let i = 0; i < 5; i++) await two.launcher.tick();
|
|
assert.deepEqual(two.runtime.sessionClears("acme").map((c) => c.stage), ["drained"]);
|
|
assert.deepEqual(events(f, "session-clear-stop"), []);
|
|
assert.equal(startTimeOf(runner.pid), runnerStartTime, "the survivor wasn't signalled");
|
|
|
|
runner.kill("SIGKILL");
|
|
await gone;
|
|
const done = await settle(() => {
|
|
const e = two.call("session.operation", { thread: A, op: "clear-1" });
|
|
return e.start === "started" && e;
|
|
}, 30000);
|
|
assert.deepEqual([done.state, done.start], ["retired", "started"]);
|
|
assert.deepEqual(events(f, "session-clear-stop"), []);
|
|
});
|
|
|
|
// D7 step 3: a drained run that doesn't exit within the stop timeout. The ticks are the test's, so
|
|
// its runner is stopped from outside its namespace before the clear's one TERM, as
|
|
// launcher.test.mjs's SIGTERM test does, and the TERM waits. The clear records uncertain once,
|
|
// sends nothing more, and retires when the run's end comes.
|
|
test("REQ-THREAD-2 T4 D7: a drained run that outlives the stop timeout makes the clear uncertain once, with one signal, and it retires on the run's end", { skip, timeout: 90000 }, async (t) => {
|
|
const prepared = prepare(t);
|
|
const { f } = prepared;
|
|
const host = await start(t, prepared, { tickMs: 3600000, stopTimeoutMs: 1000 });
|
|
const tick = () => host.launcher.tick();
|
|
const { A, D } = thread(host.call);
|
|
const first = send(host.call, A, D, "before the clear");
|
|
await tick();
|
|
await first.reply();
|
|
const [s] = readSessions(f.dataRoot);
|
|
const kill = () => {
|
|
if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGKILL");
|
|
};
|
|
t.after(kill);
|
|
host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" });
|
|
// The run's own receive drains it; no tick is needed for that.
|
|
const drained = await settle(() => host.runtime.sessionClears("acme").find((c) => c.stage === "drained"));
|
|
assert.equal(drained?.run, s.run);
|
|
process.kill(s.runnerPid, "SIGSTOP");
|
|
await tick();
|
|
assert.equal(events(f, "session-clear-stop").length, 1);
|
|
const uncertain = await settle(async () => {
|
|
await tick();
|
|
return host.runtime.sessionClears("acme").find((c) => c.stage === "uncertain");
|
|
}, 15000);
|
|
assert.equal(uncertain?.run, s.run);
|
|
for (let i = 0; i < 5; i++) await tick();
|
|
assert.equal(events(f, "session-clear-stop").length, 1);
|
|
assert.equal(events(f, "session-clear-uncertain").length, 1);
|
|
assert.equal(startTimeOf(s.runnerPid), s.runnerStartTime, "nothing forced it");
|
|
process.kill(s.runnerPid, "SIGCONT");
|
|
const done = await settle(async () => {
|
|
await tick();
|
|
const e = host.call("session.operation", { thread: A, op: "clear-1" });
|
|
return e.start === "started" && e;
|
|
}, 30000);
|
|
assert.deepEqual([done.state, done.start], ["retired", "started"]);
|
|
assert.equal(events(f, "session-clear-stop").length, 1);
|
|
assert.equal(events(f, "session-clear-uncertain").length, 1);
|
|
});
|
|
|
|
// T6 (sage-3001): the bus that proves the stop is configured but unavailable. Its host's op
|
|
// rejects as host.mjs does when the broker exits before it replies (broker-exited), first at the
|
|
// exit proof, then at retirement. While it is down nothing retires, no successor launches and the
|
|
// cleared session's slot isn't handed on; each failure is logged once, and the clear resumes where
|
|
// it stopped once the bus answers again.
|
|
test("REQ-THREAD-2 T4 T6: a bus unavailable at the exit proof and then at retirement holds the clear with no retirement, successor or launch, and it resumes when the bus is back", { skip, timeout: 90000 }, async (t) => {
|
|
const prepared = prepare(t);
|
|
const { f } = prepared;
|
|
const host = await start(t, prepared);
|
|
const down = new Set();
|
|
const op = host.host.op;
|
|
host.host.op = (m) => (down.has(m.op) ? Promise.reject(Object.assign(new Error("broker exited before it replied"), { code: "broker-exited" })) : op(m));
|
|
const { A, D } = thread(host.call);
|
|
await send(host.call, A, D, "before the clear").reply();
|
|
const stage = () => host.runtime.sessionClears("acme").map((c) => c.stage);
|
|
const lines = () => host.logs.filter((l) => l.startsWith("clear clear-1") || l.startsWith("thread tick:"));
|
|
const line = `clear clear-1 (session ${D}): broker-exited`;
|
|
const held = () => {
|
|
assert.deepEqual(events(f, "session-clear-retired"), []);
|
|
assert.deepEqual(events(f, "session-clear-failed"), []);
|
|
assert.deepEqual(events(f, "launch").map((e) => e.session), [D]);
|
|
assert.deepEqual(readSessions(f.dataRoot), [], "nothing runs in the cleared session's slot");
|
|
assert.deepEqual(host.call("session.list", { thread: A }).map((r) => [r.id, r.default]), [[D, true]]);
|
|
};
|
|
|
|
down.add("sessionClearExited");
|
|
const S = host.call("session.clear", { thread: A, session: D, revision: 0, confirm: true, op: "clear-1" }).successor;
|
|
// The drained run is stopped and ends; the exit proof is what fails.
|
|
await settle(() => events(f, "end").length === 1 && host.logs.includes(line));
|
|
await new Promise((r) => setTimeout(r, 1000));
|
|
assert.deepEqual(lines(), [line]);
|
|
assert.deepEqual(stage(), ["drained"]);
|
|
assert.equal(events(f, "session-clear-stop").length, 1);
|
|
held();
|
|
|
|
down.clear();
|
|
down.add("sessionClearRetire");
|
|
await settle(() => stage()[0] === "exited");
|
|
await new Promise((r) => setTimeout(r, 1000));
|
|
assert.deepEqual(stage(), ["exited"]);
|
|
assert.deepEqual(lines(), [line], "the same code isn't logged again");
|
|
held();
|
|
|
|
down.clear();
|
|
const done = await settle(() => {
|
|
const e = host.call("session.operation", { thread: A, op: "clear-1" });
|
|
return e.start === "started" && e;
|
|
}, 30000);
|
|
assert.deepEqual([done.state, done.start, done.refusal], ["retired", "started", null]);
|
|
const kinds = launchLog(f).map((e) => [e.event, e.session ?? null]).filter(([e]) => ["session-clear-stop", "end", "session-clear-retired", "launch"].includes(e));
|
|
assert.deepEqual(kinds, [
|
|
["launch", D],
|
|
["session-clear-stop", D],
|
|
["end", D],
|
|
["session-clear-retired", D],
|
|
["launch", S],
|
|
]);
|
|
assert.deepEqual(lines(), [line]);
|
|
assert.deepEqual(host.call("session.list", { thread: A }).map((r) => [r.id, r.state, r.default]), [
|
|
[D, "retired", false],
|
|
[S, "active", true],
|
|
]);
|
|
});
|