OpenAI Agents API

import http from "node:http";
import OpenAI from "openai";
import { Boxd, NotFoundError } from "@boxd-sh/sdk";

const EXECUTOR_KEY = process.env.OPENAI_EXECUTOR_API_KEY!;

const client = new OpenAI({ webhookSecret: process.env.OPENAI_WEBHOOK_SECRET });
const boxd = new Boxd();
const pendingPause = new Map<string, NodeJS.Timeout>();
const chains = new Map<string, Promise<unknown>>();

const machineName = (sessionId: string) =>
  "agent-" + sessionId.toLowerCase().replace(/[^a-z0-9-]/g, "-");
const quote = (s: string) => `'${s.replace(/'/g, `'\\''`)}'`;

// One session's machine changes state one step at a time.
function locked<T>(sessionId: string, fn: () => Promise<T>): Promise<T> {
  const next = (chains.get(sessionId) ?? Promise.resolve()).then(fn, fn);
  chains.set(sessionId, next.catch(() => {}));
  return next;
}

async function ensureMachine(sessionId: string): Promise<string> {
  const name = machineName(sessionId);
  let machine;
  try {
    machine = await boxd.machines.get(name);
  } catch (e) {
    if (!(e instanceof NotFoundError)) throw e;
    machine = await boxd.machines.create({ name, fromSnapshot: "codex-worker-base", isolated: true });
    await boxd.machines.setAutoHibernateTimeout(machine.id, 0);
  }
  if (machine.status === "suspended") await boxd.machines.resume(machine.id);   // the SDK's name for standby
  else if (machine.status === "hibernated") await boxd.machines.wake(machine.id);
  await boxd.machines.waitUntilReady(machine.id);
  return machine.id;
}

async function startExecutor(machineId: string, environment: { id: string; remote_url: string }) {
  const executor = ["codex", "exec-server", "--remote", environment.remote_url, "--environment-id", environment.id]
    .map(quote).join(" ");
  const script = `pkill -x codex; cd /workspace && exec flock -w 15 /tmp/codex-executor.lock ${executor}`;
  // env is a K=V prefix on the command line. It reaches this sh, and the executor inherits it.
  await boxd.machines.exec(machineId, {
    command: `nohup sh -c ${quote(script)} > /tmp/codex-executor.log 2>&1 &`,
    env: { CODEX_API_KEY: EXECUTOR_KEY },
  });
}

async function connect(sessionId: string) {
  cancelPause(sessionId);
  await locked(sessionId, async () => {
    const session = await client.beta.agents.sessions.retrieve(sessionId);
    if (session.environment.type !== "self_hosted") return;
    if (!session.required_actions?.some((a) => a.type === "environment_connection")) return; // a duplicate delivery
    await startExecutor(await ensureMachine(sessionId), session.environment);
  });
}

async function pauseNow(sessionId: string) {
  await locked(sessionId, async () => {
    const session = await client.beta.agents.sessions.retrieve(sessionId);
    if (session.status !== "idle" || session.required_actions?.length) return; // a turn started or input is waiting
    const machine = await boxd.machines.get(machineName(sessionId));
    await boxd.machines.exec(machine.id, { command: "pkill -x codex" }); // OpenAI sees the environment go offline
    await boxd.machines.pause(machine.id);
  });
}

function cancelPause(sessionId: string) {
  clearTimeout(pendingPause.get(sessionId));
  pendingPause.delete(sessionId);
}

async function removeMachine(sessionId: string) {
  cancelPause(sessionId);
  try {
    await boxd.machines.delete(machineName(sessionId));
  } catch (e) {
    if (!(e instanceof NotFoundError)) throw e;
  }
}

http.createServer(async (req, res) => {
  const body = await new Promise<string>((resolve) => {
    let d = ""; req.on("data", (c) => (d += c)); req.on("end", () => resolve(d));
  });
  try {
    await client.webhooks.verifySignature(body, req.headers as Record<string, string>);
  } catch {
    res.writeHead(400); return res.end("invalid signature");
  }
  const event = JSON.parse(body);
  const sessionId: string = event.data.id;
  if (event.type === "agent.session.action_required") {
    if (event.data.required_action.type === "environment_connection") {
      void connect(sessionId).catch(console.error);   // ack now, the boot outlives the delivery
    }
  } else if (event.type === "agent.session.idle") {
    cancelPause(sessionId);
    pendingPause.set(sessionId, setTimeout(() => void pauseNow(sessionId).catch(console.error), 30_000));
  } else if (event.type === "agent.session.failed") {
    void removeMachine(sessionId).catch(console.error);
  }
  res.writeHead(204); res.end();
}).listen(8000, "0.0.0.0");