sandboxArgv sets GIT_AUTHOR_NAME/EMAIL and GIT_COMMITTER_NAME/EMAIL from
the instance and business ids (name = instance id, email
<instance>@<business>.mosaic.invalid); spawnSession and the launcher pass
the ids through. checkout results carry the clone's commit (a kept clone
gets none), and the bundle prompt names it. allowed() is unchanged.
Packet agents/darkwing/work/queue-104/r1/ (manifest f99f2ae7). Dewey
approved round 1 in 28293 (posted under jarvis by mistake; 28294 is
Dewey's confirmation). Sage's gate on f00613106 plus the candidate:
every package suite and scripts/test-*.sh green (test-release failed 3
activation checks once, not reproduced; rerun 14/0).
Co-Authored-By: Claude Opus 5.5 <[email protected]>
245 lines
12 KiB
JavaScript
245 lines
12 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
|
|
// 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, 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";
|
|
|
|
export { REPO };
|
|
export const RUNNER = join(REPO, "packages", "harness", "src", "runner.mjs");
|
|
export const ADAPTERS = Object.freeze({
|
|
pi: join(REPO, "adapters", "pi", "adapter.sh"),
|
|
"claude-code": join(REPO, "adapters", "claude", "adapter.sh"),
|
|
});
|
|
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);
|
|
|
|
// The capacity family of a model: the one key of the business file's
|
|
// launch.max that the model name contains (`claude-opus-5-5` is `opus`).
|
|
// None, or more than one, is null, and the launch refuses.
|
|
export function family(model, max) {
|
|
const hits = Object.keys(max).filter((f) => model.toLowerCase().includes(f.toLowerCase()));
|
|
return hits.length === 1 ? hits[0] : null;
|
|
}
|
|
|
|
// 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, harness, brokerSocket, launchSocket,
|
|
// claudeBin; 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);
|
|
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, fam) {
|
|
return this.#sessions.filter((s) => s.business === business && s.family === fam).length;
|
|
}
|
|
running(business, instance) {
|
|
return this.#sessions.find((s) => s.business === business && s.instance === instance) ?? 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);
|
|
}
|