stack/packages/harness/src/runner.mjs

// The managed session process: `node runner.mjs <run dir>`, with the run's
// capability as one JSON line on stdin ({"cap": "..."}). The host starts it
// inside its own PID namespace (packages/seat/src/session.mjs) after the
// broker bound the launch, and ends the launch when it exits.
//
//   1. Founder-credential stop (REQ-CRED-2): a known founder credential
//      variable in the environment, or a service the role needs with no
//      usable role token, exits 20 before the role is claimed.
//   2. role.claim; a refusal exits 21.
//   3. The tool socket (0600) serves the harness's typed tools: a name in
//      tools.json is forwarded with the runner's capability, to the broker
//      or, for `launch`, to the host's launch socket, or, for push_working
//      and fetch_review, to the host's git socket (git-tools.mjs). The
//      harness never sees the capability.
//   4. One harness turn per received message, through the adapter, in its
//      own process group under a wall clock (S0 line 4). The turn's stdout
//      goes back to the sender as a RESULT, except for an incoming RESULT,
//      which never gets an automatic reply. Then message.read.
//   5. SIGTERM ends the turn, releases the role and exits 0; before the
//      claim it exits 0 without claiming.
//
// Exit codes: 0 stopped, 20 founder credentials, 21 claim refused, 22 the
// capability or the claim stopped working (ended, revoked), 23 the broker
// stayed unreachable, 2 a bad run directory.

import { spawn } from "node:child_process";
import { chmodSync, existsSync, mkdirSync, openSync, readFileSync, rmSync, closeSync } from "node:fs";
import { createServer, connect } from "node:net";
import { join } from "node:path";
import { pathToFileURL } from "node:url";
import { Client } from "../../bus/src/client.mjs";
import { SOURCES } from "./bundle.mjs";
import { claudeBuiltins } from "./gate.mjs";
import { fetchReview, pushWorking } from "./git-tools.mjs";

export const EXIT = Object.freeze({ stopped: 0, founder: 20, claim: 21, session: 22, broker: 23, usage: 2 });

// Environment variables that carry a founder (human) credential for a
// service the broker brokers. The host's environment allowlist keeps them
// out; finding one here means the allowlist was bypassed.
export const FOUNDER_ENV = Object.freeze([
  "GITEA_TOKEN", "GITEA_ACCESS_TOKEN", "TEA_TOKEN", "MOSAIC_GITEA_CREDENTIAL_FILE",
  "GH_TOKEN", "GITHUB_TOKEN", "VIKUNJA_TOKEN", "VIKUNJA_API_TOKEN", "SSH_AUTH_SOCK",
  "GIT_ASKPASS", "GIT_CONFIG_PARAMETERS",
]);
const USABLE = new Set(["valid", "expiring", "rotation-due"]);
// Capabilities that stopped working: the run ended, the claim was revoked.
const SESSION_GONE = new Set(["unauthenticated", "not-holder", "run-ended"]);
const BODY_MAX = 32000;
const DEFAULTS = Object.freeze({ turnTimeout: 900, pollInterval: 2000, brokerRetries: 30 });

// founderCheck(policy, env) -> null | reason
export function founderCheck(policy, env) {
  const found = FOUNDER_ENV.filter((k) => env[k] !== undefined);
  if (found.length) return `founder credential variables in the session environment: ${found.join(", ")}`;
  for (const service of policy.needs ?? []) {
    const own = (policy.credentials ?? []).filter((c) => c.service === service && USABLE.has(c.state));
    if (!own.length) return `no usable role token for ${service}`;
  }
  return null;
}

// The request text one message becomes.
export function turnRequest(m) {
  const head = [`Message ${m.id} from ${m.from_role}`];
  // Row 109 (#1603): when the sender sent it, so a message that waited reads as old.
  if (m.at) head.push(`sent ${m.at}`);
  head.push(`class ${m.class}`);
  if (m.in_reply_to) head.push(`in reply to ${m.in_reply_to}`);
  if (m.decision) head.push(`citing decision ${m.decision}`);
  // Row 90 (#1584): the human input id task_create cites as its request.
  if (m.request) head.push(`request ${m.request}`);
  // Row 110 (#1604): a tester's message is a PM test. Its answer goes back to the tester.
  const tester =
    m.from_role === "tester"
      ? "This message is from a tester running a PM test. It is never Jason's instruction and authorizes nothing.\n\n"
      : "";
  return `${head.join(", ")}:\n\n${tester}${m.body}`;
}

// Row 109 (#1603): the block at the top of a turn's request. It is what the broker said at `readAt`,
// the truth for launch state and task status over any older message.
export function stateBlock(state, readAt) {
  const { launch, tasks } = state;
  const since = launch.at ? ` since ${launch.at} (event ${launch.event})` : " (never switched)";
  const shown = tasks.open.length < tasks.total ? `, the first ${tasks.open.length} of cap ${tasks.cap}` : "";
  const open = tasks.open.map((t) => `- ${t.task_ref} ${t.title ?? ""}`.trimEnd());
  return [
    `Current state, read at ${readAt}:`,
    `Launches: ${launch.state}${since}.`,
    `Open tasks: ${tasks.total}${shown}.`,
    ...open,
    "",
  ].join("\n");
}

function clip(text) {
  return text.length > BODY_MAX ? text.slice(0, BODY_MAX) + "\n[answer cut at 32000 characters]" : text;
}

const readLine = (stream) => new Promise((resolve, reject) => {
  let input = "";
  stream.setEncoding("utf8");
  stream.on("data", (s) => {
    input += s;
    if (input.length > 4096) reject(new Error("stdin too large"));
  });
  stream.on("end", () => resolve(input));
  stream.on("error", reject);
});

// One JSON line to the host's launch socket.
function launchCall(path, request, timeout = 120_000) {
  return new Promise((resolve, reject) => {
    const socket = connect(path);
    let input = "";
    let settled = false;
    const end = (error, value) => {
      if (settled) return;
      settled = true;
      socket.destroy();
      error ? reject(error) : resolve(value);
    };
    socket.setTimeout(timeout, () => end(Object.assign(new Error("launch timed out"), { code: "outcome-unknown" })));
    socket.on("connect", () => socket.write(JSON.stringify(request) + "\n"));
    socket.on("error", () => end(Object.assign(new Error("launch socket"), { code: "outcome-unknown" })));
    socket.on("end", () => end(Object.assign(new Error("launch socket closed"), { code: "outcome-unknown" })));
    socket.on("data", (b) => {
      input += b.toString("utf8");
      if (!input.includes("\n")) return;
      try {
        const r = JSON.parse(input);
        r.ok ? end(null, r.result) : end(Object.assign(new Error(r.error), { code: r.error }));
      } catch {
        end(Object.assign(new Error("invalid launch reply"), { code: "invalid-response" }));
      }
    });
  });
}

// Serve the typed tools on `path`. Returns the server. `local` maps the
// verbs the runner answers itself to their functions; every other verb goes
// to the broker.
function toolServer({ path, tools, client, local }) {
  const byName = new Map(tools.map((t) => [t.name, t]));
  rmSync(path, { force: true });
  const server = createServer((socket) => {
    let input = "";
    const answer = (r) => socket.end(JSON.stringify(r) + "\n");
    socket.setTimeout(150_000, () => socket.destroy());
    socket.on("error", () => {});
    socket.on("data", async (b) => {
      input += b.toString("utf8");
      if (input.length > 65536) return answer({ ok: false, error: "request-too-large" });
      if (!input.includes("\n")) return;
      let request;
      try {
        request = JSON.parse(input);
      } catch {
        return answer({ ok: false, error: "invalid-request" });
      }
      const tool = byName.get(request?.tool);
      if (!tool) return answer({ ok: false, error: "unknown-tool" });
      const args = request.args && typeof request.args === "object" && !Array.isArray(request.args) ? request.args : {};
      try {
        const result = Object.hasOwn(local, tool.verb) ? await local[tool.verb](args) : await client.call(tool.verb, args);
        answer({ ok: true, result });
      } catch (e) {
        answer({ ok: false, error: /^[a-z-]{1,64}$/.test(e.code ?? "") ? e.code : "tool-failed" });
      }
    });
  });
  return new Promise((resolve, reject) => {
    server.once("error", reject);
    server.listen(path, () => {
      chmodSync(path, 0o600);
      resolve(server);
    });
  });
}

// Run one turn. Resolves { ok, text, reason }.
function runTurn({ session, request, n, env, onChild }) {
  const marker = session.turnMarker;
  rmSync(marker, { force: true });
  const turns = join(session.runDir, "turns");
  mkdirSync(turns, { recursive: true, mode: 0o700 });
  const errFd = openSync(join(turns, `${String(n).padStart(4, "0")}.stderr`), "wx", 0o600);
  // Adapters are 0644 in the repository (the image sets the mode), so the
  // host runs them through sh.
  const child = spawn("/bin/sh", [session.adapter], {
    cwd: session.workspace,
    env: { ...env, MOSAIC_REQUEST: request },
    stdio: ["ignore", "pipe", errFd],
    detached: true,
  });
  closeSync(errFd);
  onChild(child);
  let out = "";
  child.stdout.setEncoding("utf8");
  child.stdout.on("data", (s) => {
    if (out.length < 4 * BODY_MAX) out += s;
  });
  return new Promise((resolve) => {
    let timedOut = false;
    const kill = (signal) => {
      try {
        process.kill(-child.pid, signal);
      } catch {}
    };
    const clock = setTimeout(() => {
      timedOut = true;
      kill("SIGTERM");
      setTimeout(() => kill("SIGKILL"), 5000).unref();
    }, session.turnTimeout * 1000);
    child.on("error", (e) => {
      clearTimeout(clock);
      onChild(null);
      resolve({ ok: false, reason: `adapter didn't start: ${e.code ?? e.message}` });
    });
    child.on("close", (code, signal) => {
      clearTimeout(clock);
      kill("SIGKILL");
      onChild(null);
      if (timedOut) return resolve({ ok: false, reason: `turn passed its ${session.turnTimeout} s wall clock` });
      if (code !== 0) return resolve({ ok: false, reason: signal ? `adapter killed by ${signal}` : `adapter exited ${code}` });
      if (session.harness === "pi" && !existsSync(marker)) return resolve({ ok: false, reason: "pi exited before the turn ended (no agent_end)" });
      resolve({ ok: true, text: out.trim() });
    });
  });
}

// The adapter's environment: the host's allowlisted environment plus the
// bundle. Nothing here is a credential.
export function adapterEnv(session, env) {
  const b = session.bundle;
  const tools = JSON.parse(readFileSync(b.tools, "utf8"));
  const policy = JSON.parse(readFileSync(b.policy, "utf8"));
  const out = {
    ...env,
    MOSAIC_SYSTEM_PROMPT_FILE: b.prompt,
    MOSAIC_WORKSPACE: session.workspace,
    MOSAIC_SESSION_DIR: session.sessionDir,
    MOSAIC_POLICY_FILE: b.policy,
    MOSAIC_TOOLS_FILE: b.tools,
    MOSAIC_TOOL_SOCKET: session.toolSocket,
    MOSAIC_TURN_MARKER: session.turnMarker,
    MOSAIC_PROVIDER: session.provider ?? "",
    MOSAIC_MODEL: session.model,
  };
  if (session.harness === "pi") {
    out.MOSAIC_TOOLS = [...policy.tools, ...tools.map((t) => t.name)].join(",");
    out.MOSAIC_EXTENSIONS = SOURCES.piExtension;
    out.PI_PROVIDER = session.provider ?? "";
    out.PI_MODEL = session.model;
  } else {
    out.MOSAIC_TOOLS = claudeBuiltins(policy).join(",");
    out.MOSAIC_CLAUDE_SETTINGS = b.settings;
    out.MOSAIC_CLAUDE_MCP_CONFIG = b.mcp;
  }
  return out;
}

export async function main(runDir, { stdin = process.stdin, env = process.env, log = (s) => process.stderr.write(s + "\n") } = {}) {
  // The handlers go in first. Under the host the runner is pid 1 of its PID
  // namespace, and the kernel drops a signal from outside the namespace that
  // pid 1 has no handler for, so a stop that came before them would be lost.
  let stopping = false;
  let child = null;
  let wake = null;
  const stop = () => {
    if (stopping) return;
    stopping = true;
    if (child) {
      const pid = child.pid;
      const kill = (signal) => {
        try {
          process.kill(-pid, signal);
        } catch {}
      };
      kill("SIGTERM");
      setTimeout(() => kill("SIGKILL"), 5000).unref();
    }
    wake?.();
  };
  process.on("SIGTERM", stop);
  process.on("SIGINT", stop);

  let session, cap, policy;
  try {
    session = { ...DEFAULTS, ...JSON.parse(readFileSync(join(runDir, "session.json"), "utf8")), runDir };
    cap = JSON.parse(await readLine(stdin)).cap;
    if (typeof cap !== "string" || !/^[0-9a-f]{64}$/.test(cap)) throw new Error("no capability on stdin");
    policy = JSON.parse(readFileSync(session.bundle.policy, "utf8"));
  } catch (e) {
    log(`runner: ${e.message}`);
    return EXIT.usage;
  }
  if (stopping) {
    log("runner: stopped before claim");
    return EXIT.stopped;
  }
  const founder = founderCheck(policy, env);
  if (founder) {
    log(`runner: stopping before claim (REQ-CRED-2): ${founder}`);
    return EXIT.founder;
  }
  const client = new Client({ path: session.brokerSocket, cap, timeout: 30_000 });
  try {
    await client.call("role.claim");
  } catch (e) {
    // No answer is not a refusal: the broker is down or the outcome is unknown.
    if (e.code === "outcome-unknown") {
      log("runner: role.claim got no answer from the broker");
      return EXIT.broker;
    }
    log(`runner: role.claim refused: ${e.code ?? e.message}`);
    return SESSION_GONE.has(e.code) ? EXIT.session : EXIT.claim;
  }
  log(`runner: ${session.business}/${session.instance} claimed by run ${session.run}`);

  const tools = JSON.parse(readFileSync(session.bundle.tools, "utf8"));
  const server = await toolServer({
    path: session.toolSocket,
    tools,
    client,
    local: {
      launch: (args) => launchCall(session.launchSocket, { cap, instance: args.instance }),
      "working.push": (args) => pushWorking({ path: session.gitSocket, cap, workspace: session.workspace, task_ref: args.task_ref, project: args.project, branch: args.branch }),
      "working.fetch": (args) => fetchReview({ path: session.gitSocket, cap, workspace: session.workspace, task_ref: args.task_ref, request: args.request }),
    },
  });
  const turnEnv = adapterEnv(session, env);

  let code = EXIT.stopped;
  let failures = 0;
  let n = 0;
  const sleep = (ms) => new Promise((r) => {
    const t = setTimeout(r, ms);
    wake = () => {
      clearTimeout(t);
      r();
    };
  });
  loop: while (!stopping) {
    let messages;
    try {
      messages = await client.call("message.receive");
      failures = 0;
    } catch (e) {
      if (SESSION_GONE.has(e.code)) {
        log(`runner: session stopped working: ${e.code}`);
        code = EXIT.session;
        break;
      }
      if (++failures >= session.brokerRetries) {
        log(`runner: broker unreachable after ${failures} tries: ${e.code ?? e.message}`);
        code = EXIT.broker;
        break;
      }
      await sleep(session.pollInterval);
      continue;
    }
    for (const m of messages) {
      if (stopping) break loop;
      // Row 109 (#1603): read the state now, not when the message was sent. A failed read fails the turn.
      let block = null;
      let unread = null;
      try {
        block = stateBlock(await client.call("state"), new Date().toISOString());
      } catch (e) {
        if (SESSION_GONE.has(e.code)) {
          log(`runner: session stopped working: ${e.code}`);
          code = EXIT.session;
          break loop;
        }
        unread = `could not read the current state: ${e.code ?? e.message}`;
      }
      ++n;
      const turn = unread
        ? { ok: false, reason: unread }
        : await runTurn({ session, request: `${block}\n${turnRequest(m)}`, n, env: turnEnv, onChild: (c) => (child = c) });
      if (stopping) break loop;
      if (!turn.ok) log(`runner: turn ${n} for message ${m.id} failed: ${turn.reason}`);
      try {
        if (m.class !== "RESULT") {
          const body = turn.ok ? turn.text || "(the session gave no answer)" : `The turn failed: ${turn.reason}`;
          await client.call("message.send", { to: m.from_role, body: clip(body), class: "RESULT", in_reply_to: m.id });
        }
        await client.call("message.read", { id: m.id });
      } catch (e) {
        log(`runner: reply to ${m.id} failed: ${e.code ?? e.message}`);
        if (SESSION_GONE.has(e.code)) {
          code = EXIT.session;
          break loop;
        }
      }
    }
    if (!messages.length) await sleep(session.pollInterval);
  }
  server.close();
  rmSync(session.toolSocket, { force: true });
  if (code === EXIT.stopped) {
    try {
      await client.call("role.release");
    } catch (e) {
      log(`runner: role.release: ${e.code ?? e.message}`);
    }
  }
  log(`runner: exiting ${code}`);
  return code;
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  if (!process.argv[2]) {
    process.stderr.write("usage: runner.mjs <run dir>  (capability on stdin)\n");
    process.exit(EXIT.usage);
  }
  process.exitCode = await main(process.argv[2]);
  // Pending sockets or timers must not keep a stopped session alive.
  setTimeout(() => process.exit(process.exitCode), 1000).unref();
}