review loop: working repositories, git socket, push and review verdicts, row 93 (#1587)

Commits move as v2 bundles over bus-host/git.sock (decision 135). The bus
host keeps one bare working repository per business and project, seeds main
from the project root under a per-repository lock, fetches a pushed bundle
with fsck into a temp ref, requires history shared with main, and moves the
task's work ref only as a fast-forward. Host and runner git run with no
system or global config, hooks at /dev/null and local transports only. The
socket binds only for roles holding git.push.working or review.verdict.

The broker records a push as the action.allowed for git.push.working with
its commit, takes review.request for the latest push and one review.verdict
per request, and refuses task.close on a pushed task without an approve
verdict for the latest push. recordEvent reserves both shapes, so agents
can't forge either.

Reviewed by Dewey (28196) and Sage (28197), both approve. Gate on 7a729aef:
14 node suites and 9 scripts green.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

packages/cli/src/git-socket.mjs

@@ -0,0 +1,262 @@
+// The host's git socket (#1587), `<dataRoot>/bus-host/git.sock` (0600). A
+// coder's runner pushes its work to the business's working repository
+// through it, and a reviewer's runner fetches a commit it was asked to
+// review. The bundle is the only thing that crosses: the host never runs git
+// in a session's clone and never opens a path in a session's workspace
+// (packages/seat/src/working.mjs says why). The sandbox binds the socket only
+// for a role that holds git.push.working or review.verdict.
+//
+// Every connection starts with one JSON line of at most 4096 bytes.
+//
+//   push   {"cap", "op": "push", "task_ref", "project", "branch", "size"}
+//          The host answers {"ok": true, "send": <size>} or a refusal. On
+//          "send" the runner writes exactly <size> bytes of a v2 git bundle,
+//          and the host answers {"ok": true, "result": {ref, commit,
+//          previous}} or a refusal.
+//   fetch  {"cap", "op": "fetch", "task_ref", "request", "have": [<oid>...]}
+//          The host answers {"ok": true, "result": {project, ref, commit,
+//          size}} followed by <size> bytes of a bundle whose one ref is
+//          refs/mosaic/review, or a refusal.
+//
+// A refusal is {"ok": false, "error": "<code>"}, and the connection ends.
+// The host checks, in order: identity (a role of this business), the request
+// (task ref, project, branch, size, at most 32 MiB: bundle-too-large), then
+// the broker: git.push.working for a push, the review.requested for a fetch
+// (packages/bus/src/broker.mjs). Only then does it take any bundle bytes.
+// A push is recorded, as the action.allowed for git.push.working with its
+// commit, under the repository's lock once the ref moved; when the broker
+// refuses the record, the ref goes back.
+// The host's own refusals go through the broker's refuse(), so each is an
+// action.refused event on the task's trail; the broker records its own.
+//
+// A bundle file lives in bus-host/ (0700) under a random name, and the host
+// deletes it on every path.
+
+import { randomBytes } from "node:crypto";
+import { closeSync, createReadStream, existsSync, lstatSync, openSync, rmSync, statSync, writeSync, chmodSync } from "node:fs";
+import { createServer } from "node:net";
+import { join } from "node:path";
+import { pipeline } from "node:stream/promises";
+import { gitSocketPath } from "../../seat/src/session.mjs";
+import { BUNDLE_MAX, FSCK, HAVE_MAX, push, review, validBranch, workRef, WorkingError } from "../../seat/src/working.mjs";
+import { CliError } from "./errors.mjs";
+
+const HEADER_MAX = 4096;
+const HEADER_TIMEOUT_MS = 10000;
+// The longest the host waits between two chunks of a bundle.
+const BODY_IDLE_MS = 30000;
+// Connections served at once; more refuse git-busy.
+const MAX_ACTIVE = 8;
+const CAP = /^[0-9a-f]{64}$/;
+const OID = /^[0-9a-f]{40}([0-9a-f]{24})?$/;
+
+// A refusal the host decided; it goes to the broker's refuse().
+class Refusal extends Error {
+  constructor(code, message = code) {
+    super(message);
+    this.code = code;
+  }
+}
+// A refusal the broker decided and recorded itself.
+class Recorded extends Error {
+  constructor(code) {
+    super(code);
+    this.code = code;
+  }
+}
+
+// Resolves { req, rest } once a whole first line came; rest is what followed it.
+function readHeader(socket) {
+  return new Promise((resolve, reject) => {
+    let buf = Buffer.alloc(0);
+    const done = (err, value) => {
+      socket.off("data", onData);
+      socket.off("end", onGone);
+      socket.off("close", onGone);
+      socket.off("timeout", onGone);
+      socket.setTimeout(0);
+      socket.pause();
+      err ? reject(err) : resolve(value);
+    };
+    const onGone = () => done(new Refusal("invalid-request", "no request line"));
+    const onData = (chunk) => {
+      buf = Buffer.concat([buf, chunk]);
+      const nl = buf.indexOf(0x0a);
+      if (nl < 0 ? buf.length > HEADER_MAX : nl > HEADER_MAX) return done(new Refusal("invalid-request", "request line too long"));
+      if (nl < 0) return;
+      let req;
+      try {
+        req = JSON.parse(buf.subarray(0, nl).toString("utf8"));
+      } catch {
+        return done(new Refusal("invalid-request", "request line isn't JSON"));
+      }
+      done(null, { req, rest: buf.subarray(nl + 1) });
+    };
+    socket.on("data", onData);
+    socket.once("end", onGone);
+    socket.once("close", onGone);
+    socket.setTimeout(HEADER_TIMEOUT_MS);
+    socket.once("timeout", onGone);
+  });
+}
+
+// Exactly `size` bytes from the socket into a new file (0600).
+function readBody(socket, file, size) {
+  return new Promise((resolve, reject) => {
+    const fd = openSync(file, "wx", 0o600);
+    let got = 0;
+    let settled = false;
+    const done = (err) => {
+      if (settled) return;
+      settled = true;
+      socket.off("data", onData);
+      socket.off("end", onShort);
+      socket.off("close", onShort);
+      socket.off("timeout", onShort);
+      socket.setTimeout(0);
+      socket.pause();
+      closeSync(fd);
+      err ? reject(err) : resolve();
+    };
+    const onShort = () => done(new Refusal("invalid-request", `the bundle ended after ${got} of ${size} bytes`));
+    const onData = (chunk) => {
+      if (got + chunk.length > size) return done(new Refusal("invalid-request", "more bytes than the request's size"));
+      try {
+        writeSync(fd, chunk);
+      } catch (e) {
+        return done(e);
+      }
+      got += chunk.length;
+      if (got === size) done();
+    };
+    socket.on("data", onData);
+    socket.once("end", onShort);
+    socket.once("close", onShort);
+    socket.setTimeout(BODY_IDLE_MS);
+    socket.once("timeout", onShort);
+    socket.resume();
+  });
+}
+
+// host: startHost()'s handle; business: loadBusiness() output. bundleMax and
+// fsck are for the tests.
+export function createGitSocket({ host, business: b, dataRoot, env = process.env, log = () => {}, bundleMax = BUNDLE_MAX, fsck = FSCK }) {
+  const path = gitSocketPath(dataRoot);
+  const dir = join(dataRoot, "bus-host");
+  let server = null;
+  let closed = false;
+  const sockets = new Set();
+  const active = new Set();
+
+  const op = async (message) => {
+    try {
+      return await host.op(message);
+    } catch (e) {
+      throw new Recorded(e.code ?? "not-authorized");
+    }
+  };
+  const project = (id) => (typeof id === "string" && Object.hasOwn(b.projects, id) ? b.projects[id] : null);
+
+  async function doPush(socket, cap, req, rest, file) {
+    const { task_ref, branch, size } = req;
+    const p = project(req.project);
+    if (!p || !validBranch(branch) || !Number.isSafeInteger(size) || size < 1) throw new Refusal("invalid-request");
+    workRef(task_ref);
+    if (size > bundleMax) throw new Refusal("bundle-too-large", `${size} bytes, over ${bundleMax}`);
+    // The runner waits for "send"; bytes before it break the protocol.
+    if (rest.length) throw new Refusal("invalid-request", "bundle bytes before send");
+    await op({ op: "checkPush", cap, task_ref });
+    socket.write(`${JSON.stringify({ ok: true, send: size })}\n`);
+    await readBody(socket, file, size);
+    const result = await push({
+      dataRoot, business: b.id, project: req.project, root: p.root, taskRef: task_ref, branch, bundle: file, env, fsck,
+      record: (r) => op({ op: "recordPush", cap, push: { task_ref, project: req.project, branch, ...r } }),
+    });
+    socket.end(`${JSON.stringify({ ok: true, result })}\n`);
+  }
+
+  async function doFetch(socket, cap, req, file) {
+    const { task_ref, request, have = [] } = req;
+    if (typeof request !== "string" || !Array.isArray(have) || have.length > HAVE_MAX || have.some((h) => typeof h !== "string" || !OID.test(h))) throw new Refusal("invalid-request");
+    workRef(task_ref);
+    const r = await op({ op: "reviewFetch", cap, args: { task_ref, request } });
+    const p = project(r.project);
+    if (!p) throw new Refusal("unknown-project");
+    const made = await review({ dataRoot, business: b.id, project: r.project, root: p.root, taskRef: task_ref, commit: r.commit, have, out: file, env });
+    const size = statSync(file).size;
+    if (size > bundleMax) throw new Refusal("bundle-too-large", `${size} bytes, over ${bundleMax}`);
+    socket.write(`${JSON.stringify({ ok: true, result: { project: r.project, ref: made.ref, commit: made.commit, size } })}\n`);
+    await pipeline(createReadStream(file), socket);
+  }
+
+  async function serve(socket) {
+    let cap = null;
+    let subject = null;
+    let file = null;
+    const fail = (code) => {
+      if (!socket.destroyed) socket.end(`${JSON.stringify({ ok: false, error: code })}\n`);
+    };
+    try {
+      const { req, rest } = await readHeader(socket);
+      if (typeof req?.cap !== "string" || !CAP.test(req.cap)) return fail("unauthenticated");
+      if (closed) return fail("host-closing");
+      let who;
+      try {
+        who = await host.op({ op: "identity", cap: req.cap });
+      } catch (e) {
+        return fail(e.code ?? "unauthenticated");
+      }
+      cap = req.cap;
+      if (typeof req.task_ref === "string") subject = req.task_ref;
+      if (!who.role || who.business !== b.id) throw new Refusal("not-a-role");
+      file = join(dir, `${randomBytes(8).toString("hex")}.bundle`);
+      if (req.op === "push") await doPush(socket, cap, req, rest, file);
+      else if (req.op === "fetch") await doFetch(socket, cap, req, file);
+      else throw new Refusal("invalid-request", "op is push or fetch");
+    } catch (e) {
+      if (e instanceof Recorded) return fail(e.code);
+      const code = e instanceof Refusal || e instanceof WorkingError ? e.code : "git-failed";
+      if (code === "git-failed") log(`git socket: ${e.message}`);
+      if (cap) await host.op({ op: "refuse", cap, code, subject }).catch(() => {});
+      fail(code);
+    } finally {
+      if (file) rmSync(file, { force: true });
+    }
+  }
+
+  async function listen() {
+    if (existsSync(path)) {
+      if (!lstatSync(path).isSocket()) throw new CliError(`${path} exists and isn't a socket`, 3);
+      rmSync(path);
+    }
+    server = createServer((socket) => {
+      sockets.add(socket);
+      socket.once("close", () => sockets.delete(socket));
+      socket.on("error", () => {});
+      if (active.size >= MAX_ACTIVE) return socket.end(`${JSON.stringify({ ok: false, error: "git-busy" })}\n`);
+      const job = serve(socket).catch((e) => log(`git socket: ${e.message}`));
+      active.add(job);
+      job.finally(() => active.delete(job));
+    });
+    // bus-host/ is 0700, so the moment before the chmod exposes nothing.
+    await new Promise((resolve, reject) => {
+      server.once("error", reject);
+      server.listen(path, resolve);
+    });
+    chmodSync(path, 0o600);
+  }
+
+  // A push whose bundle already came finishes, under its lock, while the
+  // broker still runs: the launcher's close() runs before the host stops it.
+  async function close() {
+    closed = true;
+    if (!server) return;
+    const stopped = new Promise((r) => server.close(r));
+    for (const socket of sockets) socket.destroy();
+    await Promise.allSettled([...active]);
+    await stopped;
+    rmSync(path, { force: true });
+  }
+
+  return { path, listen, close };
+}

packages/cli/src/launcher.mjs

@@ -34,6 +34,10 @@
 //
 // `bus start <business> --pm` launches launch.by the same way with
 // launchedBy = the business's human and no caller capability.
+//
+// listen() also serves the git socket (src/git-socket.mjs, row 93), which
+// the sandbox binds for a role that holds git.push.working or
+// review.verdict; session.json names it as gitSocket.
 
 import { spawnSync } from "node:child_process";
 import { createServer } from "node:net";
@@ -47,6 +51,7 @@ import {
   endReason,
   family,
   hasSessionCredential,
+  gitSocketPath,
   launchSocketPath,
   newRun,
   readSessions,
@@ -61,6 +66,7 @@ import {
 } from "../../seat/src/session.mjs";
 import { BWRAP, sandboxProblem } from "../../seat/src/sandbox.mjs";
 import { CheckoutError, ensureCheckouts } from "../../seat/src/checkout.mjs";
+import { createGitSocket } from "./git-socket.mjs";
 import { cmdlineOf, startTimeOf } from "../../seat/src/proc.mjs";
 import { REPO, rolesDir } from "./config.mjs";
 import { CliError } from "./errors.mjs";
@@ -95,8 +101,9 @@ export function claudeVersion(env = process.env) {
 }
 
 // host: startHost()'s handle. tracker: true when the broker has task verbs for
-// the business. adapters, bwrap, readOnly, sessionDefaults, versions and
-// stopTimeoutMs exist for the tests; the command line never sets them.
+// the business. adapters, bwrap, readOnly, sessionDefaults, versions,
+// stopTimeoutMs and git (createGitSocket options) exist for the tests; the
+// command line never sets them.
 export function createLauncher({
   host,
   system,
@@ -108,6 +115,7 @@ export function createLauncher({
   sessionDefaults = {},
   versions = null,
   stopTimeoutMs = STOP_TIMEOUT_MS,
+  git = {},
   log = (l) => process.stderr.write(`mosaic-bus: ${l}\n`),
 }) {
   const dataRoot = system.dataRoot;
@@ -129,6 +137,7 @@ export function createLauncher({
   let closed = false;
   let server = null;
   const sockets = new Set();
+  const gitSocket = createGitSocket({ host, business: b, dataRoot, env, log, ...git });
 
   const record = (entry) => {
     try {
@@ -206,6 +215,9 @@ export function createLauncher({
       claudeVersion: v.claude,
     });
     const { manifest, ...files } = bundle.files;
+    // Row 93: the git socket, for a role that pushes work or reviews it.
+    const held = [...resolved.limits.authority.withinRole, ...resolved.limits.authority.crossRole];
+    const git = held.includes("git.push.working") || held.includes("review.verdict") ? gitSocketPath(dataRoot) : null;
     writeSessionFile(runDir, {
       ...sessionDefaults,
       business: b.id,
@@ -217,6 +229,7 @@ export function createLauncher({
       model: model.model,
       brokerSocket: host.path,
       launchSocket: launchSocketPath(dataRoot),
+      gitSocket: git,
       workspace,
       sessionDir,
       adapter: adapters[model.harness],
@@ -231,6 +244,7 @@ export function createLauncher({
       harness: model.harness,
       brokerSocket: host.path,
       launchSocket: resolved.launch ? launchSocketPath(dataRoot) : null,
+      gitSocket: git,
       claudeBin: env.MOSAIC_CLAUDE_BIN || null,
       readOnly,
     };
@@ -415,6 +429,7 @@ export function createLauncher({
       server.listen(path, resolve);
     });
     chmodSync(path, 0o600);
+    await gitSocket.listen();
   }
 
   // SIGTERM every runner, wait, then end what's left. Runs before the host
@@ -429,6 +444,7 @@ export function createLauncher({
       await stopped;
       rmSync(launchSocketPath(dataRoot), { force: true });
     }
+    await gitSocket.close();
     await chain;
     // Again every second: a runner still starting has no handler yet, and as
     // pid 1 of its namespace it never sees a signal sent before it has one.