stack/packages/seat/tests/session.test.mjs

// 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));
  // Row 68: the host's pinned Claude Code binary reaches the adapter.
  assert.equal(sessionEnv({ MOSAIC_CLAUDE_BIN: "/opt/claude" }, "r0123456789ab").MOSAIC_CLAUDE_BIN, "/opt/claude");
});

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);
});