Adds the harness plug-in kind (interface 1). packages/plugins/harness/pi and claude-code hold what core branched on: the adapter, thinking levels, preflight, version probe, sandbox mounts, builtins line, typed tool names, gate timeout, bundle files, adapter env, turn end and session-files path. Core reaches a module only through the loader. A session's route is the instance's harness, model and thinking vars, or the three fallback. vars; a missing one refuses no-route, and the system file's execution keys are never read for a session. The launcher checks the primary route, then the fallback: harness-off, harness-refused, thinking-unsupported, then preflight. A failed preflight falls back when a fallback is set, and any refusal of the fallback names the primary's failure too. Account checks run only when the route's module counts an account. The manifest records route, routeReason and harnessVersion, replacing piVersion and claudeVersion. pi gets --thinking, Claude Code --effort, both from MOSAIC_THINKING. The business file refuses a partial fallback. routes and routes.set are human-only bus verbs outside ORGANIZATION_KEYS; checkRoutes is shared by routes.set, instances.set and limits.set and checks only changed routes. The chief of staff changes an instance's model, and the assistant's whole vars (Q169), through limits.set, so its route write runs the same checks. A model outside launch.models refuses unknown-model, primary or fallback, and routes.set passes that code through. A new run binds only on a switched-on harness module name and records its transcript path, which the usage reader uses. Settings gains a Routing table with six selects per instance and loses the Role models fieldset; it posts changed rows to /api/routes. webui/src/bus.mjs is unchanged. Until Jason's `mosaic plugins set` adds "harness": ["claude-code"], a new launch refuses harness-off.
265 lines
13 KiB
JavaScript
265 lines
13 KiB
JavaScript
// Managed role sessions (slice 1 S6): where a launch lives on disk, how its
|
|
// runner starts, and how the human stops it. The bus host
|
|
// (packages/cli/src/launcher.mjs) is the only caller that starts one; it
|
|
// decides authority and capacity first, and this module only does what it
|
|
// is told.
|
|
//
|
|
// A session is `node packages/harness/src/runner.mjs <run dir>` inside the
|
|
// bubblewrap sandbox of sandbox.mjs, which sees only its own workspace, its
|
|
// run directory and the files it runs from. The runner is pid 1 of its own
|
|
// PID namespace, so a child that detaches (`setsid -f`) is reparented to the
|
|
// runner and stays under the launch's pid, which the human CLI's ancestry
|
|
// check refuses. bwrap passes the runner's exit status on, a death by signal
|
|
// as 128 + the signal; `stop` signals the runner.
|
|
//
|
|
// On disk, under the data root:
|
|
// launches/<business>/<run>/ the run directory (0700): session.json,
|
|
// bundle/, session/, turns/, runner.log,
|
|
// tools.sock, turn.done
|
|
// launches/<business>.jsonl the launch log (0600, append-only):
|
|
// every launch, refusal and end
|
|
// workspaces/<business>/<instance>/ the instance's workspace (0700), kept
|
|
// across launches
|
|
// threads/<business>/<thread>/ a thread's session store (0700), kept
|
|
// across the thread's runs (row 121):
|
|
// session/ and claude-projects/
|
|
// bus-host/sessions.json the running sessions (0600), written
|
|
// only by the host; `mosaic stop` reads it
|
|
|
|
import { spawn } from "node:child_process";
|
|
import { randomBytes } from "node:crypto";
|
|
import { appendFileSync, closeSync, existsSync, lstatSync, mkdirSync, openSync, readFileSync, renameSync, writeFileSync } from "node:fs";
|
|
import { constants } from "node:os";
|
|
import { dirname, join } from "node:path";
|
|
import { childrenOf, cmdlineOf, startTimeOf } from "./proc.mjs";
|
|
import { binds, prepareRun, REPO, sandboxArgv } from "./sandbox.mjs";
|
|
import { SeatError } from "./seat.mjs";
|
|
import { ID_PATTERN } from "../../business/src/vocabulary.mjs";
|
|
|
|
export { REPO };
|
|
export const RUNNER = join(REPO, "packages", "harness", "src", "runner.mjs");
|
|
export const SESSIONS_VERSION = 1;
|
|
|
|
export const sessionsFile = (dataRoot) => join(dataRoot, "bus-host", "sessions.json");
|
|
export const launchSocketPath = (dataRoot) => join(dataRoot, "bus-host", "launch.sock");
|
|
// Row 93 (#1587): packages/cli/src/git-socket.mjs.
|
|
export const gitSocketPath = (dataRoot) => join(dataRoot, "bus-host", "git.sock");
|
|
export const runsDir = (dataRoot, business) => join(dataRoot, "launches", business);
|
|
export const launchLogFile = (dataRoot, business) => join(dataRoot, "launches", `${business}.jsonl`);
|
|
export const workspaceDir = (dataRoot, business, instance) => join(dataRoot, "workspaces", business, instance);
|
|
|
|
// Row 121 (#1615): a thread's session store, which outlives each of the
|
|
// thread's runs so the next run resumes the harness session (REQ-THREAD-2).
|
|
// Both segments are checked before the join: an id, no "..", so no "/" and
|
|
// no leading ".". Then every directory from threads/ down must be ours, a
|
|
// real directory and private, as store.mjs safe() checks the bus. session/
|
|
// is the run's sessionDir; claude-projects/ is bound at the run's
|
|
// home/.claude/projects (sandbox.mjs binds()). Throws SeatError.
|
|
const segment = (x) => typeof x === "string" && ID_PATTERN.test(x) && !x.includes("..");
|
|
export function threadStore(dataRoot, business, thread) {
|
|
for (const [what, x] of [["business", business], ["thread", thread]]) {
|
|
if (!segment(x)) throw new SeatError(`thread store: ${what} ${JSON.stringify(x)} isn't an id`, 2);
|
|
}
|
|
const dir = join(dataRoot, "threads", business, thread);
|
|
const out = { dir, session: join(dir, "session"), projects: join(dir, "claude-projects") };
|
|
for (const d of [join(dataRoot, "threads"), join(dataRoot, "threads", business), dir, out.session, out.projects]) {
|
|
try {
|
|
mkdirSync(d, { mode: 0o700 });
|
|
} catch (e) {
|
|
if (e.code !== "EEXIST") throw e;
|
|
}
|
|
const st = lstatSync(d);
|
|
if (st.isSymbolicLink() || !st.isDirectory() || st.uid !== process.getuid() || st.mode & 0o077) throw new SeatError(`thread store: ${d} is unsafe`, 2);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
// The session's environment is an allowlist: nothing else from the host's
|
|
// environment passes, so a founder credential variable (GITEA_TOKEN,
|
|
// GH_TOKEN, SSH_AUTH_SOCK, ...) never reaches a session (REQ-CRED-2).
|
|
// Provider variables pass because the harness needs them to reach a model.
|
|
// MOSAIC_CLAUDE_BIN, the host's pinned Claude Code binary, is a path. HOME
|
|
// passes to bwrap only; inside, HOME is the run's own home/.
|
|
export const ENV_ALLOW = Object.freeze([
|
|
"HOME", "USER", "LOGNAME", "LANG", "LC_ALL", "TZ", "TERM",
|
|
"ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_BASE_URL", "CLAUDE_CODE_OAUTH_TOKEN",
|
|
"OPENAI_API_KEY", "OPENROUTER_API_KEY",
|
|
"MOSAIC_CLAUDE_BIN",
|
|
]);
|
|
|
|
// The ways a Claude Code session can log in, each a set of variables that
|
|
// must all be set; ENV_ALLOW passes them on. The bus unit loads them
|
|
// (packages/cli/README.md, "Unit"). The session gets nothing from ~/.claude,
|
|
// which the sandbox doesn't show.
|
|
export const SESSION_CREDENTIALS = Object.freeze([["CLAUDE_CODE_OAUTH_TOKEN"], ["ANTHROPIC_API_KEY"], ["ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_BASE_URL"]].map((set) => Object.freeze(set)));
|
|
export const hasSessionCredential = (env) => SESSION_CREDENTIALS.some((set) => set.every((k) => Boolean(env[k])));
|
|
|
|
export function sessionEnv(env, run) {
|
|
const out = {};
|
|
for (const k of ENV_ALLOW) if (env[k] !== undefined) out[k] = env[k];
|
|
out.PATH = [join(REPO, "node_modules", ".bin"), env.PATH ?? "/usr/bin:/bin"].join(":");
|
|
out.MOSAIC_RUN_ID = run;
|
|
return out;
|
|
}
|
|
|
|
// A new run id and its directory. Ids are short because the run directory
|
|
// holds a Unix socket (tools.sock), and socket paths stop at 107 bytes.
|
|
export function newRun(dataRoot, business) {
|
|
const root = runsDir(dataRoot, business);
|
|
mkdirSync(root, { recursive: true, mode: 0o700 });
|
|
for (let i = 0; i < 5; i++) {
|
|
const run = `r${randomBytes(6).toString("hex")}`;
|
|
const runDir = join(root, run);
|
|
try {
|
|
mkdirSync(runDir, { mode: 0o700 });
|
|
} catch (e) {
|
|
if (e.code === "EEXIST") continue;
|
|
throw e;
|
|
}
|
|
if (Buffer.byteLength(join(runDir, "tools.sock")) > 107) throw new SeatError(`data root path too long for a socket: ${runDir}`, 2);
|
|
return { run, runDir };
|
|
}
|
|
throw new SeatError("could not pick a free run id", 1);
|
|
}
|
|
|
|
export function writeSessionFile(runDir, session) {
|
|
writeFileSync(join(runDir, "session.json"), `${JSON.stringify(session, null, 2)}\n`, { mode: 0o600, flag: "wx" });
|
|
}
|
|
|
|
export function appendLaunchLog(dataRoot, business, entry) {
|
|
const file = launchLogFile(dataRoot, business);
|
|
mkdirSync(dirname(file), { recursive: true, mode: 0o700 });
|
|
appendFileSync(file, `${JSON.stringify({ at: new Date().toISOString(), ...entry })}\n`, { mode: 0o600 });
|
|
}
|
|
|
|
// Start the runner in its sandbox. stdin stays open for the capability line
|
|
// the host writes once the broker has bound the launch; stdout and stderr go
|
|
// to runner.log. paths: what sandbox.mjs's binds() takes besides the run
|
|
// directory (workspace, adapter, harnessMounts, brokerSocket, launchSocket,
|
|
// gitSocket, threadStore; readOnly for tests). business and role (the
|
|
// instance) give the session its git identity. `bwrap` is for the tests.
|
|
export function spawnSession({ runDir, env, paths, business, role, node = process.execPath, bwrap }) {
|
|
prepareRun(runDir, { threadStore: paths.threadStore ?? null });
|
|
const argv = sandboxArgv(binds({ ...paths, runDir, node }), runDir, [node, RUNNER, runDir], { business, role, bwrap });
|
|
const out = openSync(join(runDir, "runner.log"), "a", 0o600);
|
|
try {
|
|
return spawn(argv[0], argv.slice(1), { cwd: runDir, env, stdio: ["pipe", out, out], detached: true });
|
|
} finally {
|
|
closeSync(out);
|
|
}
|
|
}
|
|
|
|
// A runner's command line is `<node> <RUNNER> <run dir>`. Matching argv[1]
|
|
// rather than any argument skips bwrap's forked child before its exec, whose
|
|
// argv still names the runner.
|
|
export const isRunner = (pid) => (cmdlineOf(pid) ?? [])[1] === RUNNER;
|
|
|
|
// The runner's pid and start time as seen from here: the outer bwrap
|
|
// process's only child.
|
|
export async function runnerOf(child, { timeoutMs = 5000 } = {}) {
|
|
const deadline = Date.now() + timeoutMs;
|
|
while (Date.now() < deadline) {
|
|
if (child.exitCode !== null || child.signalCode !== null) break;
|
|
const pid = childrenOf(child.pid)[0];
|
|
const start = pid ? startTimeOf(pid) : null;
|
|
if (start && isRunner(pid)) return { pid, startTime: start };
|
|
await new Promise((r) => setTimeout(r, 20));
|
|
}
|
|
throw new SeatError(`runner for ${child.pid} did not start`, 1);
|
|
}
|
|
|
|
// The outer bwrap's exit as the runner's: bwrap exits 128 + n when the
|
|
// runner dies of signal n, and the runner's own codes are all below 128.
|
|
const SIGNALS = Object.fromEntries(Object.entries(constants.signals).map(([name, n]) => [n, name]));
|
|
export function sessionExit(code, signal) {
|
|
if (!signal && code > 128 && SIGNALS[code - 128]) return { code: null, signal: SIGNALS[code - 128] };
|
|
return { code, signal };
|
|
}
|
|
|
|
// The launch end reason the broker records for a runner's exit
|
|
// (packages/harness/README.md, "The runner").
|
|
const REASONS = { 0: "stopped", 2: "failed", 20: "founder-credentials", 21: "claim-refused", 22: "session-invalid", 23: "broker-unreachable" };
|
|
export function endReason(code, signal) {
|
|
if (signal) return "killed";
|
|
return REASONS[code] ?? "failed";
|
|
}
|
|
|
|
// The host's view of its running sessions, kept in memory and mirrored to
|
|
// sessions.json for `mosaic stop` and `mosaic launches list`.
|
|
export class Registry {
|
|
#dataRoot;
|
|
#sessions = [];
|
|
constructor(dataRoot) {
|
|
this.#dataRoot = dataRoot;
|
|
this.#write();
|
|
}
|
|
get sessions() {
|
|
return this.#sessions.map((s) => ({ ...s }));
|
|
}
|
|
add(entry) {
|
|
this.#sessions.push({ ...entry });
|
|
this.#write();
|
|
}
|
|
remove(business, run) {
|
|
this.#sessions = this.#sessions.filter((s) => !(s.business === business && s.run === run));
|
|
this.#write();
|
|
}
|
|
count(business, provider, model = undefined) {
|
|
return this.#sessions.filter((s) => s.business === business && s.provider === provider && (model === undefined || s.model === model)).length;
|
|
}
|
|
// A thread instance runs once per thread (row 121), so `thread` is part of
|
|
// the key; null is a session that holds no thread.
|
|
running(business, instance, thread = null) {
|
|
return this.#sessions.find((s) => s.business === business && s.instance === instance && (s.thread ?? null) === thread) ?? null;
|
|
}
|
|
#write() {
|
|
const file = sessionsFile(this.#dataRoot);
|
|
mkdirSync(dirname(file), { recursive: true, mode: 0o700 });
|
|
const tmp = `${file}.${process.pid}.tmp`;
|
|
writeFileSync(tmp, `${JSON.stringify({ sessionsVersion: SESSIONS_VERSION, sessions: this.#sessions }, null, 2)}\n`, { mode: 0o600 });
|
|
renameSync(tmp, file);
|
|
}
|
|
}
|
|
|
|
// sessions.json as `mosaic stop` sees it. `live` is true only when the
|
|
// runner still runs with the recorded start time.
|
|
export function readSessions(dataRoot) {
|
|
const file = sessionsFile(dataRoot);
|
|
if (!existsSync(file)) return [];
|
|
let doc;
|
|
try {
|
|
doc = JSON.parse(readFileSync(file, "utf8"));
|
|
} catch {
|
|
throw new SeatError(`sessions file is unreadable: ${file}`, 2);
|
|
}
|
|
if (doc?.sessionsVersion !== SESSIONS_VERSION || !Array.isArray(doc.sessions)) throw new SeatError(`sessions file is malformed: ${file}`, 2);
|
|
return doc.sessions.map((s) => ({ ...s, live: startTimeOf(s.runnerPid) === s.runnerStartTime }));
|
|
}
|
|
|
|
// `mosaic stop <run>`: SIGTERM to the run's runner, after checking that the
|
|
// pid still has the recorded start time and runs the runner. Same-UID
|
|
// process control, not an authority check. The runner releases its role
|
|
// and the host records the end.
|
|
export async function stopSession(dataRoot, run, { timeoutMs = 30000 } = {}) {
|
|
const hits = readSessions(dataRoot).filter((s) => s.run === run);
|
|
if (hits.length === 0) throw new SeatError(`no running session has run id ${run}; see mosaic launches list`, 2);
|
|
const s = hits[0];
|
|
if (!s.live) return { stopped: false, reason: `run ${run} is not running (stale entry)` };
|
|
if (!isRunner(s.runnerPid)) throw new SeatError(`pid ${s.runnerPid} is not a session runner; refusing to signal it`, 2);
|
|
// Sent again every second: a runner that is still starting has no handler
|
|
// yet, and as pid 1 of its namespace it never sees a signal sent before then.
|
|
const deadline = Date.now() + timeoutMs;
|
|
let next = 0;
|
|
while (Date.now() < deadline) {
|
|
if (startTimeOf(s.runnerPid) !== s.runnerStartTime) return { stopped: true, business: s.business, instance: s.instance, run };
|
|
if (Date.now() >= next) {
|
|
try {
|
|
process.kill(s.runnerPid, "SIGTERM");
|
|
} catch {}
|
|
next = Date.now() + 1000;
|
|
}
|
|
await new Promise((r) => setTimeout(r, 100));
|
|
}
|
|
throw new SeatError(`run ${run} did not stop within ${Math.round(timeoutMs / 1000)} s`, 1);
|
|
}
|