git mv packages/business -> packages/organization (@mosaic/organization). The old package stays as one-line re-export shims of each moved module, with src/business.mjs re-exporting src/organization-config.mjs and src/cli.mjs forwarding to the moved executable. BusinessError is OrganizationConfigError (one constructor, name kept BusinessError); OrganizationError stays distinct. The bus adapter gains organization-context.mjs (busOrganization), with business.mjs (busBusiness) as the same function. Import consumers, manifests, the lockfile, scripts/mosaic, stack-copy.sh COPY_PATHS (both roots), seat sandbox RUNNER_FILES and the copy closure walker (organization-imports.mjs, both roots required) move to the new path. v1 serialization, v1 writers and the stage A fixtures are unchanged; the stage A contract tests stay where they were and pass as before.
206 lines
10 KiB
JavaScript
206 lines
10 KiB
JavaScript
// Row 302 (#1809, Vikunja 588): the idle stop releases a thread run's claim in the threadIdle op
|
|
// and sends SIGTERM only after the op returns. A runner call the broker answers in between used to
|
|
// get not-holder, and the run ended session-invalid/22 instead of stopped/0. This test holds the op
|
|
// open until a call has landed in that window and the runner has exited, so it doesn't depend on
|
|
// timing. The launcher skips a tick while one is running (launcher.mjs, the thread pass), so nothing
|
|
// else signals a run while the op is held.
|
|
//
|
|
// The host is in process, as in thread-launch.test.mjs, with the same fixture.
|
|
|
|
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 } 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;
|
|
|
|
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);
|
|
}
|
|
|
|
function prepare(t) {
|
|
const root = tmp(t, "mosaic-idle-race-");
|
|
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 = 1;
|
|
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 };
|
|
}
|
|
|
|
// thread-launch.test.mjs's start(), with threadIdle replaced by the caller.
|
|
async function start(t, { f, adapter }, threadIdle) {
|
|
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 = {
|
|
'request.admission': m => runtime.admission(m),
|
|
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) => threadIdle(runtime, m),
|
|
mayLaunch: (m) => runtime.mayLaunch(m),
|
|
landHeld: (m) => runtime.landHeld(m.record),
|
|
landHolders: (m) => runtime.landHolders(m.business),
|
|
landRelease: (m) => runtime.landRelease(m.record),
|
|
};
|
|
const refusing = (what, fn) => async (m) => {
|
|
try {
|
|
return await 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 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: 100,
|
|
log: () => {},
|
|
});
|
|
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 { runtime, launcher, call, close };
|
|
}
|
|
|
|
const launchLog = (f) =>
|
|
existsSync(launchLogFile(f.dataRoot, "acme"))
|
|
? readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l))
|
|
: [];
|
|
const runnerLog = (f, run) => {
|
|
const p = join(f.dataRoot, "launches", "acme", run, "runner.log");
|
|
return existsSync(p) ? readFileSync(p, "utf8") : "";
|
|
};
|
|
const ended = (f, run) => launchLog(f).find((e) => e.event === "end" && e.run === run);
|
|
// The runner's own verdict on a refusal: the launcher-release line, or a lost session's.
|
|
const verdict = (f, run) => /runner: (session released by the launcher|session stopped working: [a-z-]+)\n/.exec(runnerLog(f, run))?.[1] ?? null;
|
|
const settled = async (fn) => {
|
|
try {
|
|
return await until(fn);
|
|
} catch {
|
|
return null;
|
|
}
|
|
};
|
|
|
|
async function ask(call, thread, body) {
|
|
const sent = call("thread.send", { thread, body });
|
|
await until(() => call("thread.messages", { thread }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT"));
|
|
}
|
|
|
|
test("row 302 REQ-THREAD-2 an idle-released run whose call lands before the SIGTERM ends stopped/0; a run whose claim the human revoked in the same window still ends session-invalid/22", { skip, timeout: 120000 }, async (t) => {
|
|
const prepared = prepare(t);
|
|
const { f } = prepared;
|
|
// The first run the idle stop releases. Inside its held window, the human stops the other
|
|
// thread's run, which no launcher pass can signal while the op is held.
|
|
let window = null;
|
|
let host = null;
|
|
const threadIdle = async (runtime, m) => {
|
|
const r = runtime.threadIdle(m.record);
|
|
if (!r.released || window) return r;
|
|
window = { run: m.record.run, thread: m.record.thread };
|
|
const other = runtime.threadsState("acme").find((x) => x.thread !== m.record.thread && x.holder);
|
|
window.control = other ? { run: other.holder, stopped: host.call("assistant.stop", { instance: "assistant" }).stopped } : null;
|
|
// Both runners decide on their own before the op returns and the launcher can signal.
|
|
window.idle = await settled(() => verdict(f, window.run) && ended(f, window.run) && verdict(f, window.run));
|
|
window.revoked = other ? await settled(() => verdict(f, other.holder) && ended(f, other.holder) && verdict(f, other.holder)) : null;
|
|
return r;
|
|
};
|
|
host = await start(t, prepared, threadIdle);
|
|
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;
|
|
await Promise.all([ask(host.call, A, "for A"), ask(host.call, B, "for B")]);
|
|
// The op sets window.revoked last, so this waits for the whole window (dewey-0878 N2).
|
|
await until(() => window && "revoked" in window, 60000);
|
|
assert.ok(window.control, "the other thread's run held B when A's first release landed");
|
|
await until(() => ended(f, window.run) && ended(f, window.control.run), 60000);
|
|
|
|
const { run, control } = window;
|
|
assert.deepEqual(control.stopped.map((s) => s.run), [control.run], "the human's stop took only the other thread's run");
|
|
// The released run refused in the window and exited by itself before the op returned.
|
|
assert.equal(window.idle, "session released by the launcher", runnerLog(f, run));
|
|
const log = launchLog(f);
|
|
const end = (r) => log.find((e) => e.event === "end" && e.run === r);
|
|
assert.deepEqual([end(run).reason, end(run).exitCode, end(run).signal], ["stopped", 0, null]);
|
|
assert.ok(log.indexOf(end(run)) < log.findIndex((e) => e.event === "thread-idle" && e.run === run), "the end came before the idle stop's signal");
|
|
assert.doesNotMatch(runnerLog(f, run), /session stopped working/);
|
|
assert.match(runnerLog(f, run), /runner: exiting 0\n/);
|
|
// The launcher already released the claim, so the runner doesn't release it at exit (dewey-0878 N1).
|
|
assert.doesNotMatch(runnerLog(f, run), /runner: role\.release:/);
|
|
// The control: a real not-holder in the same window still ends 22, with no signal.
|
|
assert.equal(window.revoked, "session stopped working: not-holder", runnerLog(f, control.run));
|
|
assert.deepEqual([end(control.run).reason, end(control.run).exitCode, end(control.run).signal], ["session-invalid", 22, null]);
|
|
assert.match(runnerLog(f, control.run), /runner: exiting 22\n/);
|
|
|
|
await host.close();
|
|
const store = new Store(f.dataRoot);
|
|
t.after(() => store.close());
|
|
const codes = (r) => store.all("SELECT body FROM events WHERE kind='action.refused' AND actor_run=? ORDER BY seq", r).map((e) => JSON.parse(e.body).code);
|
|
assert.ok(codes(run).includes("launcher-released"), JSON.stringify(codes(run)));
|
|
assert.ok(!codes(run).includes("not-holder"), JSON.stringify(codes(run)));
|
|
assert.ok(codes(control.run).includes("not-holder"), JSON.stringify(codes(control.run)));
|
|
});
|