// Managed role sessions on disk and as processes (src/session.mjs). The // runner is the real one, under the real unshare, against an in-process // broker on a real socket; the harness turn is packages/harness's fake // adapter. The launcher's checks are tested in packages/cli. import { test } from "node:test"; import assert from "node:assert/strict"; import { spawnSync } from "node:child_process"; import { existsSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { Broker } from "../../bus/src/broker.mjs"; import { serve } from "../../bus/src/server.mjs"; import { Store } from "../../bus/src/store.mjs"; import { cmdlineOf, startTimeOf } from "../src/proc.mjs"; import { SeatError } from "../src/seat.mjs"; import { appendLaunchLog, endReason, ENV_ALLOW, family, isRunner, launchLogFile, newRun, readSessions, Registry, runnerOf, sessionEnv, sessionsFile, spawnSession, stopSession, UNSHARE, writeSessionFile, } from "../src/session.mjs"; const FAKE = new URL("../../harness/tests/fixtures/fake-adapter.mjs", import.meta.url).pathname; const NAMESPACES = spawnSync(UNSHARE[0], [...UNSHARE.slice(1), "true"]).status === 0; const skip = NAMESPACES ? false : "unprivileged user and PID namespaces are unavailable here"; function scratch(t) { const dir = realpathSync(mkdtempSync(join(tmpdir(), "mosaic-session-"))); t.after(() => rmSync(dir, { recursive: true, force: true })); return dir; } const mode = (p) => statSync(p).mode & 0o777; test("family: exactly one launch.max key in the model name, else null", () => { const max = { opus: 2, sonnet: 4 }; assert.equal(family("claude-opus-5-5", max), "opus"); assert.equal(family("Claude-Sonnet-5-5", max), "sonnet"); assert.equal(family("glm-5.3-flash", max), null); assert.equal(family("opus-sonnet-mix", max), null); }); test("sessionEnv passes only the allowlist, the repo's bin on PATH, and the run id", () => { const env = sessionEnv({ HOME: "/h", ANTHROPIC_API_KEY: "k", GITEA_TOKEN: "t", GH_TOKEN: "t", SSH_AUTH_SOCK: "/s", MOSAIC_CONFIG: "/c", PATH: "/usr/bin" }, "r0123456789ab"); assert.deepEqual(Object.keys(env).sort(), ["ANTHROPIC_API_KEY", "HOME", "MOSAIC_RUN_ID", "PATH"]); assert.match(env.PATH, /node_modules\/\.bin:\/usr\/bin$/); assert.equal(env.MOSAIC_RUN_ID, "r0123456789ab"); for (const k of ["GITEA_TOKEN", "GH_TOKEN", "SSH_AUTH_SOCK"]) assert.ok(!ENV_ALLOW.includes(k)); }); test("newRun: short ids, 0700 directories, and a refusal when the socket path won't fit", (t) => { const root = scratch(t); const a = newRun(join(root, "data"), "acme"); const b = newRun(join(root, "data"), "acme"); assert.match(a.run, /^r[0-9a-f]{12}$/); assert.notEqual(a.run, b.run); assert.equal(mode(a.runDir), 0o700); assert.equal(mode(join(root, "data", "launches", "acme")), 0o700); const deep = join(root, "x".repeat(100)); assert.throws(() => newRun(deep, "acme"), (e) => e instanceof SeatError && /too long for a socket/.test(e.message)); }); test("session file and launch log: 0600, the session file written once", (t) => { const root = scratch(t); const { runDir } = newRun(root, "acme"); writeSessionFile(runDir, { run: "x" }); assert.equal(mode(join(runDir, "session.json")), 0o600); assert.throws(() => writeSessionFile(runDir, { run: "y" }), /EEXIST/); appendLaunchLog(root, "acme", { event: "launch", run: "x" }); appendLaunchLog(root, "acme", { event: "end", run: "x" }); const file = launchLogFile(root, "acme"); assert.equal(mode(file), 0o600); const lines = readFileSync(file, "utf8").trim().split("\n").map((l) => JSON.parse(l)); assert.deepEqual(lines.map((l) => l.event), ["launch", "end"]); assert.ok(lines.every((l) => !Number.isNaN(Date.parse(l.at)))); }); test("endReason maps the runner's exit codes; a signal is killed", () => { assert.deepEqual( [0, 2, 20, 21, 22, 23, 7].map((c) => endReason(c, null)), ["stopped", "failed", "founder-credentials", "claim-refused", "session-invalid", "broker-unreachable", "failed"], ); assert.equal(endReason(null, "SIGKILL"), "killed"); }); test("Registry mirrors to sessions.json; readSessions marks live entries; bad files refuse", (t) => { const root = scratch(t); const r = new Registry(root); assert.equal(mode(sessionsFile(root)), 0o600); assert.deepEqual(readSessions(root), []); const me = { business: "acme", instance: "coder", run: "r1", family: "sonnet", runnerPid: process.pid, runnerStartTime: startTimeOf(process.pid) }; r.add(me); r.add({ ...me, instance: "pm", run: "r2", family: "opus", runnerStartTime: "1" }); assert.equal(r.count("acme", "sonnet"), 1); assert.equal(r.count("other", "sonnet"), 0); assert.equal(r.running("acme", "coder").run, "r1"); assert.deepEqual(readSessions(root).map((s) => [s.run, s.live]), [["r1", true], ["r2", false]]); r.remove("acme", "r1"); assert.equal(r.running("acme", "coder"), null); assert.deepEqual(readSessions(root).map((s) => s.run), ["r2"]); writeFileSync(sessionsFile(root), "{"); assert.throws(() => readSessions(root), (e) => e instanceof SeatError && /unreadable/.test(e.message)); writeFileSync(sessionsFile(root), JSON.stringify({ sessionsVersion: 9, sessions: [] })); assert.throws(() => readSessions(root), (e) => e instanceof SeatError && /malformed/.test(e.message)); }); test("stopSession refuses an unknown run, reports a stale one, and won't signal a pid that isn't a runner", async (t) => { const root = scratch(t); const r = new Registry(root); await assert.rejects(stopSession(root, "nope"), /no running session has run id nope/); r.add({ business: "acme", instance: "coder", run: "stale", runnerPid: process.pid, runnerStartTime: "1" }); assert.deepEqual(await stopSession(root, "stale"), { stopped: false, reason: "run stale is not running (stale entry)" }); r.add({ business: "acme", instance: "pm", run: "me", runnerPid: process.pid, runnerStartTime: startTimeOf(process.pid) }); await assert.rejects(stopSession(root, "me"), /is not a session runner; refusing to signal it/); }); test("a session runs under unshare as pid 1 of its namespace, claims, answers, and stops on mosaic stop", { skip }, async (t) => { const root = scratch(t); const store = new Store(root); const broker = new Broker({ store, businesses: { acme: { id: "acme", human: "jason", arbiters: { technical: "pm", delivery: "pm" }, roles: { pm: { authority: { withinRole: ["message.send"], crossRole: [] } }, coder: { authority: { withinRole: ["message.send"], crossRole: [] } }, }, }, }, }); const brokerSocket = join(root, "bus.sock"); const server = await serve({ broker, path: brokerSocket }); t.after(async () => { await server.close().catch(() => {}); store.close(); }); const dataRoot = join(root, "data"); const { run, runDir } = newRun(dataRoot, "acme"); const bundle = join(runDir, "bundle"); const ws = join(root, "ws"); mkdirSync(bundle, { mode: 0o700 }); mkdirSync(ws); writeFileSync(join(bundle, "prompt.md"), "# test\n"); writeFileSync(join(bundle, "policy.json"), JSON.stringify({ harness: "pi", workspace: ws, tools: [], typed: [], network: "none", needs: [], credentials: [] })); writeFileSync(join(bundle, "tools.json"), "[]"); const adapter = join(root, "adapter.sh"); writeFileSync(adapter, `exec '${process.execPath}' '${FAKE}'\n`, { mode: 0o644 }); writeSessionFile(runDir, { business: "acme", instance: "coder", run, launchedBy: "pm", harness: "pi", provider: "fake", model: "fake-model", brokerSocket, launchSocket: join(root, "launch.sock"), workspace: ws, sessionDir: join(runDir, "session"), adapter, bundle: { prompt: join(bundle, "prompt.md"), policy: join(bundle, "policy.json"), tools: join(bundle, "tools.json") }, toolSocket: join(runDir, "tools.sock"), turnMarker: join(runDir, "turn.done"), pollInterval: 50, }); const child = spawnSession({ runDir, env: sessionEnv({ HOME: process.env.HOME, PATH: process.env.PATH, GITEA_TOKEN: "x" }, run) }); t.after(() => { try { process.kill(child.pid, "SIGKILL"); } catch {} }); const exited = new Promise((r) => child.once("close", (code, signal) => r({ code, signal }))); const runner = await runnerOf(child); assert.notEqual(runner.pid, child.pid); assert.deepEqual(cmdlineOf(child.pid).slice(0, UNSHARE.length), UNSHARE); assert.ok(isRunner(runner.pid)); assert.equal(isRunner(child.pid), false, "unshare's argv names the runner too"); // Inside its namespace the runner is pid 1. const nspid = readFileSync(`/proc/${runner.pid}/status`, "utf8").match(/^NSpid:\s*(.*)$/m)[1].trim().split(/\s+/); assert.equal(nspid.at(-1), "1"); const environ = readFileSync(`/proc/${runner.pid}/environ`, "utf8").split("\0"); assert.ok(environ.includes(`MOSAIC_RUN_ID=${run}`)); assert.ok(!environ.some((v) => v.startsWith("GITEA_TOKEN="))); const cap = broker.bindLaunch({ business: "acme", role: "coder", run, harness: "pi", pid: child.pid, startTime: startTimeOf(child.pid) }); const pm = broker.bindLaunch({ business: "acme", role: "pm", run: "pm-run", harness: "pi" }); broker.request(pm, { verb: "role.claim" }); child.stdin.end(`${JSON.stringify({ cap })}\n`); const sent = broker.request(pm, { verb: "message.send", args: { to: "coder", body: "hello", class: "REQUEST" } }); let reply; const end = Date.now() + 15000; while (!reply && Date.now() < end) { reply = broker.request(pm, { verb: "message.receive" }).find((m) => m.in_reply_to === sent.id); if (!reply) await new Promise((r) => setTimeout(r, 50)); } assert.ok(reply, "the session answered"); assert.match(reply.body, /^echo: Message .* from pm, class REQUEST:\n\nhello/); const registry = new Registry(dataRoot); registry.add({ business: "acme", instance: "coder", run, pid: child.pid, startTime: startTimeOf(child.pid), runnerPid: runner.pid, runnerStartTime: runner.startTime }); const stopped = await stopSession(dataRoot, run, { timeoutMs: 10000 }); assert.deepEqual(stopped, { stopped: true, business: "acme", instance: "coder", run }); assert.deepEqual(await exited, { code: 0, signal: null }, "unshare passes the runner's exit 0 on"); assert.equal(store.get("SELECT op FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").op, "release"); assert.equal(existsSync(join(runDir, "tools.sock")), false); assert.equal(mode(join(runDir, "runner.log")), 0o600); });