// 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 ` inside // `unshare --user --map-current-user --pid --fork --kill-child --mount-proc`. // 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. unshare itself ignores // SIGTERM and passes the runner's exit status on; `stop` signals the runner. // // On disk, under the data root: // launches/// the run directory (0700): session.json, // bundle/, session/, turns/, runner.log, // tools.sock, turn.done // launches/.jsonl the launch log (0600, append-only): // every launch, refusal and end // workspaces/// 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 { dirname, join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { childrenOf, cmdlineOf, startTimeOf } from "./proc.mjs"; import { SeatError } from "./seat.mjs"; export const REPO = resolve(dirname(fileURLToPath(import.meta.url)), "..", "..", ".."); 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 UNSHARE = Object.freeze(["unshare", "--user", "--map-current-user", "--pid", "--fork", "--kill-child", "--mount-proc"]); 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"); 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. 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", ]); 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. stdin stays open for the capability line the host writes // once the broker has bound the launch; stdout and stderr go to runner.log. // `namespace: false` exists for hosts without unprivileged user namespaces // in the tests only; the host always passes true. export function spawnSession({ runDir, env, namespace = true, node = process.execPath }) { const out = openSync(join(runDir, "runner.log"), "a", 0o600); const argv = [...(namespace ? UNSHARE : []), node, RUNNER, runDir]; 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 ` `. Matching argv[1] // rather than any argument skips unshare'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. Under unshare the // runner is the outer process's only child; without a namespace it is the // process itself. export async function runnerOf(child, { namespace = true, timeoutMs = 5000 } = {}) { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { if (child.exitCode !== null || child.signalCode !== null) break; const pid = namespace ? childrenOf(child.pid)[0] : child.pid; 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 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 `: 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); }