Review Composer agent adapter setup

This maintained Node.js example serves Millwork’s async_job_v1 agent protocol. Millwork creates a branch, calls this adapter with a stable attempt identity, and later reads the branch head from its GitHub App. The adapter checks out that branch, runs a coding agent, commits its work, and pushes without force. Millwork does not trust the SHA returned by the adapter as final artifact proof.

Copy the adapter

Copy this complete file into server.mjs on your agent host. Mintlify’s code block has a copy button. The setup below explains the required Git access, token, and HTTPS ingress.

server.mjs

import { createHash, randomUUID, timingSafeEqual } from "node:crypto";
import { spawn, spawnSync } from "node:child_process";
import { createServer } from "node:http";
import { mkdir, readFile, rename, writeFile } from "node:fs/promises";
import { basename, join, resolve } from "node:path";

const SHA = /^[0-9a-f]{40}$/;
const ID = /^[A-Za-z0-9_-]{1,200}$/;
const BRANCH = /^refs\/heads\/millwork\/[A-Za-z0-9/_-]{1,160}$/;
const REPOSITORY = /^[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+$/;
const MAX_BODY = 16_384;

function required(value, name) {
  if (!value || typeof value !== "string") throw new Error(`Set ${name}`);
  return value;
}

export function adapterConfig(env = process.env) {
  const repository = required(env.REVIEW_REPOSITORY, "REVIEW_REPOSITORY");
  if (!REPOSITORY.test(repository)) throw new Error("REVIEW_REPOSITORY must be owner/repo");
  const bearerToken = required(env.REVIEW_ADAPTER_TOKEN, "REVIEW_ADAPTER_TOKEN");
  if (bearerToken.length < 32) throw new Error("REVIEW_ADAPTER_TOKEN must have at least 32 characters");
  const remoteUrl = required(env.REVIEW_REMOTE_URL, "REVIEW_REMOTE_URL");
  if (!/^git@github\.com:[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+\.git$/.test(remoteUrl)
    && !/^https:\/\/github\.com\/[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+\.git$/.test(remoteUrl)) {
    throw new Error("REVIEW_REMOTE_URL must be a GitHub SSH or credential-helper HTTPS URL without a secret");
  }
  const remoteRepository = remoteUrl.replace(/^git@github\.com:/, "").replace(/^https:\/\/github\.com\//, "").replace(/\.git$/, "");
  if (remoteRepository.toLowerCase() !== repository.toLowerCase()) throw new Error("REVIEW_REMOTE_URL must match REVIEW_REPOSITORY");
  const workRoot = resolve(required(env.REVIEW_WORK_ROOT, "REVIEW_WORK_ROOT"));
  const agentArgv = JSON.parse(env.REVIEW_AGENT_ARGV_JSON ?? '["codex","exec","--approve-for-me","-"]');
  if (!Array.isArray(agentArgv) || !agentArgv.length || agentArgv.some((arg) => typeof arg !== "string" || !arg)) {
    throw new Error("REVIEW_AGENT_ARGV_JSON must be a nonempty JSON argv array");
  }
  return { repository, bearerToken, remoteUrl, workRoot, agentArgv };
}

/** Catch incompatible Codex flags before accepting a paid agent start. */
export function assertAgentCommandReady(config, probe = spawnSync) {
  const [binary, ...args] = config.agentArgv;
  if (basename(binary) !== "codex") return;
  if (args[0] !== "exec" || args.at(-1) !== "-") {
    throw new Error("Codex agent command must use codex exec and read its prompt from stdin (-)");
  }
  const result = probe(binary, [...args.slice(0, -1), "--help"], {
    encoding: "utf8", timeout: 10_000, maxBuffer: 64 * 1024,
    stdio: ["ignore", "pipe", "pipe"],
  });
  if (result.error || result.status !== 0) {
    const detail = String(result.stderr ?? "").trim().split("\n")[0].slice(0, 160);
    throw new Error(`Codex agent command preflight failed${detail ? `: ${detail}` : ""}`);
  }
}

/** Refuse jobs when this checkout cannot authenticate to its configured repo. */
export function assertGitRemoteReady(config, probe = spawnSync) {
  const result = probe("git", ["ls-remote", "--heads", config.remoteUrl], {
    encoding: "utf8", timeout: 20_000, maxBuffer: 64 * 1024,
    stdio: ["ignore", "pipe", "pipe"],
  });
  if (result.error || result.status !== 0 || !/^[0-9a-f]{40}\s+refs\/heads\//m.test(String(result.stdout ?? ""))) {
    throw new Error("Git remote preflight failed; check repository access and the credential helper");
  }
}

function run(argv, options = {}) {
  return new Promise((resolveRun, rejectRun) => {
    // The adapter bearer token authenticates Millwork to this server. Neither
    // the coding agent nor any git helper it launches needs that authority.
    const childEnv = { ...process.env, GIT_TERMINAL_PROMPT: "0", ...options.env };
    delete childEnv.REVIEW_ADAPTER_TOKEN;
    const child = spawn(argv[0], argv.slice(1), {
      cwd: options.cwd,
      env: childEnv,
      stdio: [options.input === undefined ? "ignore" : "pipe", "ignore", "ignore"],
      shell: false,
      signal: options.signal,
    });
    if (options.input !== undefined) {
      child.stdin.on("error", () => { /* a failed child may close stdin early */ });
      child.stdin.end(options.input);
    }
    child.on("error", rejectRun);
    child.on("exit", (code, signal) => {
      if (code === 0) resolveRun();
      else rejectRun(new Error(`${argv[0]} exited ${code ?? signal}`));
    });
  });
}

function validStart(input, config) {
  if (input.protocol !== "async_job_v1" || input.operation !== "start"
    || !ID.test(input.execution_id) || !ID.test(input.job_attempt_id) || !ID.test(input.idempotency_key)
    || input.repository !== config.repository || !SHA.test(input.base_sha)
    || !BRANCH.test(input.branch) || input.branch !== `refs/heads/millwork/run/${input.execution_id}`
    || typeof input.base_ref !== "string"
    || !/^refs\/heads\/[A-Za-z0-9/_-]{1,160}$/.test(input.base_ref)
    || !Number.isFinite(Date.parse(input.deadline_at)) || Date.parse(input.deadline_at) <= Date.now()
    || typeof input.task?.objective !== "string" || !input.task.objective.trim()
    || input.task.objective.length > 4_000) {
    throw new Error("invalid start identity");
  }
}

function jobId(attemptId, idempotencyKey) {
  return `job_${createHash("sha256").update(`${attemptId}\0${idempotencyKey}`).digest("hex").slice(0, 32)}`;
}

function equalToken(actual, expected) {
  const one = Buffer.from(actual ?? "");
  const two = Buffer.from(expected);
  return one.length === two.length && timingSafeEqual(one, two);
}

function comparableStart(input) {
  const { protocol, operation, execution_id, job_attempt_id, idempotency_key, repository,
    base_ref, base_sha, branch, deadline_at, task } = input;
  return { protocol, operation, execution_id, job_attempt_id, idempotency_key, repository,
    base_ref, base_sha, branch, deadline_at, task };
}

export class AsyncJobAdapter {
  constructor(config, command = run) {
    this.config = config;
    this.command = command;
    this.jobs = new Map();
    this.controllers = new Map();
    this.startLocks = new Map();
  }

  path(id) { return join(this.config.workRoot, "jobs", `${id}.json`); }

  async persist(job) {
    await mkdir(join(this.config.workRoot, "jobs"), { recursive: true, mode: 0o700 });
    const path = this.path(job.job_id);
    const temp = `${path}.${process.pid}.${randomUUID()}.tmp`;
    await writeFile(temp, JSON.stringify(job), { mode: 0o600 });
    await rename(temp, path);
  }

  async load(id) {
    if (this.jobs.has(id)) return this.jobs.get(id);
    let job;
    try { job = JSON.parse(await readFile(this.path(id), "utf8")); }
    catch (error) { if (error.code === "ENOENT") return null; throw error; }
    if (job.state === "queued" || job.state === "running") {
      // A process restart has lost supervision of an agent attempt. Never
      // dispatch it again under the same logical job identity.
      job.state = "failed";
      job.failure = "adapter_restarted_during_attempt";
      await this.persist(job);
    }
    this.jobs.set(id, job);
    return job;
  }

  async start(input) {
    validStart(input, this.config);
    const id = jobId(input.job_attempt_id, input.idempotency_key);
    const pending = this.startLocks.get(id);
    if (pending) {
      await pending;
      return this.start(input);
    }
    const dispatch = (async () => {
      const previous = await this.load(id);
      if (previous) {
        if (JSON.stringify(previous.start) !== JSON.stringify(comparableStart(input))) {
          throw new Error("logical job identity reused with different start request");
        }
        return { operation: "start", job_id: id };
      }
      const job = { job_id: id, start: comparableStart(input), state: "queued", usage_usd: 0 };
      await this.persist(job);
      this.jobs.set(id, job);
      const controller = new AbortController();
      this.controllers.set(id, controller);
      void this.execute(job, controller).finally(() => this.controllers.delete(id));
      return { operation: "start", job_id: id };
    })();
    this.startLocks.set(id, dispatch);
    try { return await dispatch; }
    finally { this.startLocks.delete(id); }
  }

  async execute(job, controller) {
    const { start } = job;
    const worktree = join(this.config.workRoot, "worktrees", job.job_id);
    const deadlineMs = Date.parse(start.deadline_at) - Date.now();
    const timer = setTimeout(() => controller.abort(), Math.max(1, deadlineMs));
    try {
      job.state = "running";
      await this.persist(job);
      await mkdir(join(this.config.workRoot, "worktrees"), { recursive: true, mode: 0o700 });
      await this.command(["git", "clone", "--no-checkout", this.config.remoteUrl, worktree], { signal: controller.signal });
      await this.command(["git", "-C", worktree, "fetch", "origin", start.branch], { signal: controller.signal });
      const fetched = await this.capture(["git", "-C", worktree, "rev-parse", "FETCH_HEAD"], controller.signal);
      if (fetched !== start.base_sha) throw new Error("Millwork run branch did not start at accepted base SHA");
      await this.command(["git", "-C", worktree, "checkout", "-B", start.branch.slice("refs/heads/".length), fetched], { signal: controller.signal });
      const prompt = `${start.task.objective}\n\nWork only in this checkout. Commit no secret. Do not push; the adapter will commit and push after you finish.`;
      await this.command(this.config.agentArgv, { cwd: worktree, signal: controller.signal, input: prompt });
      await this.command(["git", "-C", worktree, "add", "-A"], { signal: controller.signal });
      await this.command(["git", "-C", worktree, "-c", "user.name=Millwork Agent", "-c", "user.email=agent@getmillwork.dev", "commit", "--allow-empty", "-m", `Complete Millwork run ${start.execution_id}`], { signal: controller.signal });
      const sha = await this.capture(["git", "-C", worktree, "rev-parse", "HEAD"], controller.signal);
      if (!SHA.test(sha)) throw new Error("final SHA unavailable");
      await this.command(["git", "-C", worktree, "push", "origin", `HEAD:${start.branch}`], { signal: controller.signal });
      if (controller.signal.aborted) throw new Error("cancelled after push");
      const completed = { ...job, state: "completed", claimed_sha: sha };
      await this.persist(completed);
      Object.assign(job, completed);
    } catch (error) {
      const failed = { ...job, state: controller.signal.aborted ? "cancelled" : "failed",
        failure: String(error.message ?? error).slice(0, 200) };
      await this.persist(failed);
      Object.assign(job, failed);
    } finally { clearTimeout(timer); }
  }

  async capture(argv, signal) {
    return new Promise((resolveCapture, rejectCapture) => {
      const child = spawn(argv[0], argv.slice(1), { stdio: ["ignore", "pipe", "ignore"], shell: false, signal });
      let out = "";
      child.stdout.on("data", (chunk) => { out += chunk; if (out.length > 256) child.kill(); });
      child.on("error", rejectCapture);
      child.on("exit", (code) => code === 0 ? resolveCapture(out.trim()) : rejectCapture(new Error(`${argv[0]} read failed`)));
    });
  }

  async identify(input) {
    if (input.protocol !== "async_job_v1" || !ID.test(input.job_attempt_id)
      || !ID.test(input.idempotency_key) || (input.job_id !== null && input.job_id !== undefined && !ID.test(input.job_id))) {
      throw new Error("invalid job identity");
    }
    const id = jobId(input.job_attempt_id, input.idempotency_key);
    if (input.job_id && input.job_id !== id) throw new Error("job ID mismatch");
    const job = await this.load(id);
    if (!job) throw new Error("job not found");
    return job;
  }

  async status(input) {
    const job = await this.identify(input);
    return { operation: "status", job_id: job.job_id, state: job.state,
      ...(job.state === "completed" ? { claimed_sha: job.claimed_sha, usage_usd: job.usage_usd } : {}),
      ...(["failed", "cancelled"].includes(job.state) ? { usage_usd: job.usage_usd } : {}) };
  }

  async cancel(input) {
    const job = await this.identify(input);
    this.controllers.get(job.job_id)?.abort();
    return { operation: "cancel", accepted: true };
  }

  async handle(input) {
    if (input?.operation === "start") return this.start(input);
    if (input?.operation === "status") return this.status(input);
    if (input?.operation === "cancel") return this.cancel(input);
    throw new Error("unknown operation");
  }
}

export function createAdapterServer(adapter) {
  return createServer(async (request, response) => {
    const send = (status, body) => { response.writeHead(status, { "content-type": "application/json", "cache-control": "no-store" }); response.end(JSON.stringify(body)); };
    if (!equalToken(request.headers.authorization?.replace(/^Bearer /, ""), adapter.config.bearerToken)) {
      send(401, { error: "unauthorized" }); return;
    }
    if (request.method === "HEAD") { response.writeHead(200); response.end(); return; }
    if (request.method !== "POST") { send(405, { error: "method_not_allowed" }); return; }
    let raw = "";
    for await (const chunk of request) {
      raw += chunk;
      if (raw.length > MAX_BODY) { send(413, { error: "body_too_large" }); return; }
    }
    try { send(200, await adapter.handle(JSON.parse(raw))); }
    catch (error) { send(400, { error: String(error.message ?? error).slice(0, 200) }); }
  });
}

if (process.argv[1] && import.meta.url === new URL(`file://${resolve(process.argv[1])}`).href) {
  const config = adapterConfig();
  assertAgentCommandReady(config);
  assertGitRemoteReady(config);
  const adapter = new AsyncJobAdapter(config);
  const port = Number(process.env.REVIEW_ADAPTER_PORT ?? 8731);
  if (!Number.isInteger(port) || port < 1 || port > 65535) throw new Error("Invalid REVIEW_ADAPTER_PORT");
  createAdapterServer(adapter).listen(port, "127.0.0.1", () => {
    process.stdout.write(`Review Composer agent adapter listening on localhost:${port}\n`);
  });
}

Prepare

Use Node.js 20 or later, Git, and a working coding agent. The default command is codex exec --approve-for-me -; sign in to Codex on the adapter host first. The host needs Git access to the selected repository through SSH or its credential helper. Grant only the repository it will edit. Set up an HTTPS ingress that forwards Millwork’s authenticated HEAD and POST requests to this local process. Set these variables on the adapter host, never in a Millwork task or receipt:

export REVIEW_REPOSITORY="team/repo"
export REVIEW_REMOTE_URL="git@github.com:team/repo.git"
export REVIEW_WORK_ROOT="/var/lib/millwork-review-agent"
export REVIEW_ADAPTER_TOKEN="<random secret, at least 32 characters>"
export REVIEW_ADAPTER_PORT="8731"
node server.mjs

REVIEW_AGENT_ARGV_JSON optionally replaces the agent command with a JSON argv array such as ["codex","exec","--approve-for-me","-"]. At startup the adapter checks that a configured Codex command accepts its flags, before any agent job can start. The adapter sends the objective on standard input, not in process arguments or a shell command. A different configured agent command must read its prompt from standard input. Keep the token in your secret manager. Connect your repository in Millwork settings, then register this endpoint with POST /v1/review-composer/agents and endpoint.auth_ref: "". The response includes the agent ID. Start the private handoff with POST /v1/review-composer/agents/{arm_id}/connection/intents and a stop_choice, then open its continue_url in your signed-in browser. Enter the same token you set on this host. Read GET /v1/review-composer/agents/{arm_id}/connection for the pending cred_a_ handle and captured_generation; send both to POST /v1/review-composer/agents/{arm_id}/connection/test. A successful authenticated HEAD makes the agent ready. See the step-by-step guide for request bodies and SDK methods. An existing agent registered with an operator-provisioned cred_ handle cannot use this browser handoff: its connection-intent request returns a conflict. Register a new Review Composer agent with an empty auth_ref and follow the private handoff above. Do not send the token in registration or run requests. The adapter binds to 127.0.0.1; your ingress owns TLS.

Protocol and recovery

The adapter accepts start, status, and cancel messages at one URL. start returns a durable job_id before agent work begins. A retry with the same job_attempt_id and idempotency_key returns that ID and cannot launch another attempt. Changing the request under the same identity is rejected. The job record is persisted under REVIEW_WORK_ROOT/jobs; protect this directory and back it up according to your retention policy. Run one adapter instance per work root. The protocol’s job not found answer is reserved for a start that was never accepted. Persist the job record before starting agent work. Once accepted, status must find the same job after a process restart, even when the restart turns an interrupted attempt into a terminal failure. Millwork reissues start with the original key only after that explicit answer. If the process restarts while an attempt is active, the adapter reports failed when that job is inspected. It does not silently restart the agent. A failed push, moved branch, or timeout likewise reports failure. Millwork reads the remote branch itself before registering an artifact. Its final GitHub SHA, not the adapter’s claimed SHA, is what review sources evaluate. The protocol currently requires usage_usd. This example reports 0 because the Codex CLI does not provide a verified per-job bill to this adapter. Your agent provider’s invoice remains separate; Millwork records every attempted start and its reported usage rather than treating zero as proof of no cost.

Test without an external call

In a checkout of the source example, run:

npm test

The tests create a temporary local bare Git repository, run a fixture agent, verify the pushed final commit, replay and restart behavior, and exercise the authenticated HTTP endpoint. No GitHub, CodeRabbit, Codex, or Millwork call is made.

Get started

Build

Build safely

SDK and MCP

Review Composer agent adapter setup

Copy the adapter

Prepare

Protocol and recovery

Test without an external call

​Copy the adapter

​Prepare

​Protocol and recovery

​Test without an external call

Copy the adapter

Prepare

Protocol and recovery

Test without an external call