AgentHub/server/lib/codex-ingest.js

/**
 * @file Codex stage ingestion. Codex emits no hooks, so the path that infers a
 * lane's stage (server/routes/hooks.js → stage-detect.detect) can never see a
 * `codex` lane — and it is deliberately pinned to `claude`, because a Claude
 * hook must never touch a codex lane sharing the same directory. Codex's
 * on-disk equivalent is the rollout journal it appends live, one file per
 * thread:
 *
 *   $CODEX_HOME/sessions/<YYYY>/<MM>/<DD>/rollout-<ts>-<threadId>.jsonl
 *     line 1: {"type":"session_meta","payload":{"cwd":…,"session_id":…}}
 *     then:   {"type":"response_item","payload":{"type":"custom_tool_call",…}}
 *
 * This module tails those journals for the cwd of every `codex` lane, maps each
 * tool-call payload onto the same `{tool_name, tool_input}` shape a Claude hook
 * delivers, and feeds the EXISTING detect() + recordDetection() path — so a
 * codex lane paints the same amber "detected" badge under the same rules
 * (forward-only, declared-wins) and never renders `done`. Like
 * workflow-ingest.js, everything here is fail-safe: a malformed or partial
 * journal throws only locally and is skipped, and a poll pass never throws.
 *
 * Codex 0.149 also mirrors threads into version-suffixed SQLite files
 * (`state_5.sqlite`, `thread_history_1.sqlite`), but every thread row still
 * points at a live rollout file, so reading the JSONL costs no coupling to a
 * schema the next Codex release renames. `rolloutFiles()` is the one function
 * to re-point if that ever stops being true.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

const fs = require("node:fs");
const path = require("node:path");
const os = require("node:os");

// `*** Update File: <path>` lines inside an apply_patch heredoc.
const PATCH_FILE_RE = /^\*\*\* (Add|Update|Delete) File: (.+)$/gm;
const SHELL_NAMES = new Set(["exec", "exec_command", "shell", "local_shell", "container.exec"]);

const CODEX_ACTIVE_MS = 120_000; // a journal written within 2 min means the agent is working
const TAIL_BYTES = 64 * 1024; // first sight of a live file: read only its recent tail
const MAX_FILES = 40; // newest-first cap so a years-deep sessions/ tree can't stall a pass
const MAX_READ_BYTES = 4 * 1024 * 1024; // one pass never buffers more than this per journal
const MAX_DEPTH = 4; // sessions/<YYYY>/<MM>/<DD>/<file>

/** The `cmd` of a shell call, whether it arrived as JSON arguments or an exec script. */
function commandFromPayload(payload, text) {
  try {
    const parsed = JSON.parse(payload.arguments);
    if (parsed && typeof parsed.cmd === "string") return parsed.cmd;
    if (parsed && Array.isArray(parsed.command)) return parsed.command.join(" ");
  } catch {
    /* not JSON — an exec script; the detect rules match inside it verbatim */
  }
  return text;
}

/**
 * Hook-shaped events for one rollout payload. A patch script yields one event
 * per touched file (Write for a new file, Edit otherwise — the split the
 * pipeline templates' detect rules assume); a shell call yields one Bash event.
 * Anything else — reasoning, messages, tool OUTPUT, a web search — yields
 * nothing, so an unknown Codex tool can never be mistaken for work.
 *
 * @param {object|null} payload - One rollout line's `payload` object.
 * @returns {Array<{tool_name: string, tool_input: object}>}
 */
function eventsFromPayload(payload) {
  if (!payload || typeof payload !== "object") return [];
  if (payload.type !== "custom_tool_call" && payload.type !== "function_call") return [];
  const name = typeof payload.name === "string" ? payload.name : "";
  const text = typeof payload.input === "string" ? payload.input : String(payload.arguments || "");

  if (name === "update_plan") return [{ tool_name: "TaskCreate", tool_input: {} }];
  if (name === "spawn_agent") {
    let taskName = "";
    try {
      taskName = JSON.parse(payload.arguments).task_name || "";
    } catch {
      /* encrypted or malformed arguments — the tool name alone still matters */
    }
    return [{ tool_name: "Agent", tool_input: { subagent_type: taskName } }];
  }
  if (!SHELL_NAMES.has(name)) return [];

  // A patch heredoc reaches us inside a JS string literal, so its line breaks
  // are usually a literal backslash-n rather than newlines (real rollouts carry
  // both forms in the same script). Unescape before scanning or the file list
  // comes back empty and a whole edit looks like a plain shell command.
  const patched = [...text.replace(/\\n/g, "\n").matchAll(PATCH_FILE_RE)];
  if (patched.length) {
    return patched.map(([, verb, file]) => ({
      tool_name: verb === "Add" ? "Write" : "Edit",
      tool_input: { file_path: file.trim() },
    }));
  }
  return [{ tool_name: "Bash", tool_input: { command: commandFromPayload(payload, text) } }];
}

function codexHome() {
  return process.env.CODEX_HOME || path.join(os.homedir(), ".codex");
}

/** Every rollout journal under sessions/, newest mtime first, capped at MAX_FILES. */
function rolloutFiles() {
  const out = [];
  const walk = (dir, depth) => {
    let entries = [];
    try {
      entries = fs.readdirSync(dir, { withFileTypes: true });
    } catch {
      return; // absent or unreadable — Codex may simply not be installed
    }
    for (const entry of entries) {
      const full = path.join(dir, entry.name);
      if (entry.isDirectory()) {
        if (depth < MAX_DEPTH) walk(full, depth + 1);
      } else if (entry.isFile() && /^rollout-.*\.jsonl$/.test(entry.name)) {
        try {
          const stat = fs.statSync(full);
          out.push({ path: full, mtime: stat.mtimeMs, size: stat.size });
        } catch {
          /* vanished mid-scan */
        }
      }
    }
  };
  walk(path.join(codexHome(), "sessions"), 0);
  return out.sort((a, b) => b.mtime - a.mtime).slice(0, MAX_FILES);
}

const cwdCache = new Map(); // rollout path → thread cwd (written once, at thread creation)

/** The thread's working directory, from the session_meta on the file's first line. */
function rolloutCwd(file) {
  if (cwdCache.has(file)) return cwdCache.get(file);
  let cwd = null;
  try {
    const fd = fs.openSync(file, "r");
    try {
      const buf = Buffer.alloc(64 * 1024);
      const read = fs.readSync(fd, buf, 0, buf.length, 0);
      const head = buf.toString("utf8", 0, read).split("\n")[0];
      cwd = JSON.parse(head).payload.cwd || null;
    } finally {
      fs.closeSync(fd);
    }
  } catch {
    cwd = null;
  }
  if (typeof cwd !== "string" || !cwd) cwd = null;
  cwdCache.set(file, cwd);
  return cwd;
}

const offsets = new Map(); // rollout path → bytes already ingested

/** Test seam: forget every journal's ingest position (and cached cwd). */
function __resetOffsets() {
  offsets.clear();
  cwdCache.clear();
}

// Both maps are keyed by file path, so in a process that runs for months they
// would otherwise hold one dead entry per journal Codex ever wrote. Pruned only
// once well past the per-pass scan size, so ordinary churn keeps its positions
// and a journal that drops out of the scan behaves exactly as it does after a
// server restart: re-seen as new, and read only if it is being written to.
const MAX_TRACKED = MAX_FILES * 4;

function pruneTracking(seen) {
  for (const map of [offsets, cwdCache]) {
    if (map.size <= MAX_TRACKED) continue;
    for (const key of map.keys()) if (!seen.has(key)) map.delete(key);
  }
}

/** Read `[from, size)` of a file, or an empty buffer when it can't be read. */
function readRange(file, from, size) {
  try {
    const fd = fs.openSync(file, "r");
    try {
      const buf = Buffer.alloc(size - from);
      const read = fs.readSync(fd, buf, 0, buf.length, from);
      return buf.subarray(0, read);
    } finally {
      fs.closeSync(fd);
    }
  } catch {
    return Buffer.alloc(0);
  }
}

/**
 * The COMPLETE lines in `[from, size)`, plus the offset to resume from.
 *
 * Two fragments have to be handled, and only one of them is dropped:
 * - a tail read (`firstSight` with `from > 0`) starts mid-line, so its leading
 *   fragment is discarded;
 * - a read that catches the writer mid-line ends with a fragment, so the
 *   returned offset stops at the last newline and that fragment is re-read
 *   whole on the next pass.
 *
 * Every non-tail read therefore starts exactly on a line boundary, where the
 * first line is a WHOLE record. Dropping it there — the obvious "the offset
 * always lands mid-line" shortcut — loses every append that arrives one line at
 * a time, which is how a journal actually grows.
 *
 * All of it is done on the BUFFER, never on a decoded string: a tail offset can
 * land inside a multi-byte UTF-8 sequence, where the decoder emits a 3-byte
 * U+FFFD for 1-2 consumed bytes. Measuring a decoded slice with
 * `Buffer.byteLength` would then move the offset off the real line boundary and
 * every later record would fail to parse.
 */
const NEWLINE = 0x0a;

function completeLines(file, from, size, firstSight) {
  let chunk = readRange(file, from, size);
  let start = from;
  if (firstSight && from > 0) {
    const firstBreak = chunk.indexOf(NEWLINE);
    if (firstBreak === -1) return { lines: [], next: size };
    start += firstBreak + 1;
    chunk = chunk.subarray(firstBreak + 1);
  }
  const lastBreak = chunk.lastIndexOf(NEWLINE);
  if (lastBreak === -1) return { lines: [], next: start };
  return {
    lines: chunk.subarray(0, lastBreak).toString("utf8").split("\n"),
    next: start + lastBreak + 1,
  };
}

/**
 * One poll pass. Reads only the bytes appended since the last pass; on the
 * first sight of a file, only its recent tail, and nothing at all when the file
 * is already idle — replaying a thread that finished days ago would paint
 * stages from work nobody is doing.
 *
 * Detection uses the LAST matching event of the batch, mirroring "one hook, one
 * detection"; recordDetection is forward-only, so ordering inside a batch can
 * never drag a lane backwards.
 *
 * @param {{broadcast?: (laneId: number) => void}} [options]
 * @returns {number} How many lanes actually changed (and were broadcast).
 */
function ingestCodexOnce({ broadcast } = {}) {
  const lanesLib = require("./lanes");
  const { detect } = require("./stage-detect");
  const { getPipeline } = require("./pipelines");

  let codexLanes = [];
  try {
    codexLanes = lanesLib.listLanes().filter((l) => l.tool === "codex");
  } catch {
    return 0;
  }
  if (!codexLanes.length) return 0;

  const touched = new Set();
  const files = rolloutFiles();
  pruneTracking(new Set(files.map((f) => f.path)));
  for (const file of files) {
    const cwd = rolloutCwd(file.path);
    if (!cwd) continue;
    const lane = codexLanes
      .filter((l) => cwd === l.cwd || cwd.startsWith(`${l.cwd}/`))
      .sort((a, b) => b.cwd.length - a.cwd.length)[0];
    if (!lane) continue;

    const firstSight = !offsets.has(file.path);
    // An mtime in the FUTURE (clock stepped back, NTP correction, a restore that
    // preserved timestamps) makes the raw difference negative, which reads as
    // "written milliseconds ago" and would let a finished thread's journal
    // replay as live work. A file that claims to be from the future is not
    // evidence of an agent working now, so treat it as idle.
    const age = Date.now() - file.mtime;
    const active = age >= 0 && age < CODEX_ACTIVE_MS;
    let from = firstSight ? Math.max(0, file.size - TAIL_BYTES) : offsets.get(file.path);
    // A stored offset past EOF means the file was truncated or a new thread took
    // the same path. Restart from a bounded tail, never from 0: reading a
    // multi-gigabyte journal whole would allocate it all in one buffer.
    let tailOnly = firstSight;
    if (from > file.size || file.size - from > MAX_READ_BYTES) {
      from = Math.max(0, file.size - TAIL_BYTES);
      tailOnly = true;
    }

    let changed = false;
    // A journal seen for the first time while already idle is history, not work:
    // note where it ends and read nothing.
    if (firstSight && !active) {
      offsets.set(file.path, file.size);
    } else if (file.size > from) {
      const { lines, next } = completeLines(file.path, from, file.size, tailOnly);
      offsets.set(file.path, next);
      const pipeline = getPipeline(lane.pipeline);
      let detection = null;
      for (const raw of lines) {
        if (!raw.trim()) continue;
        let payload = null;
        try {
          payload = JSON.parse(raw).payload;
        } catch {
          continue; // not a record this module understands
        }
        for (const event of eventsFromPayload(payload)) {
          const hit = detect(pipeline, event);
          if (hit) detection = hit;
        }
      }
      if (detection) {
        try {
          changed = lanesLib.recordDetection(lane.id, detection).written;
        } catch {
          /* lane deleted mid-pass, or sqlite busy — an inference is never worth throwing for */
        }
      }
    }

    // Status mirror, under the same guard hook ingestion uses: only idle↔running,
    // and only for a lane the dashboard did not launch — provisioning and failed
    // are lifecycle states this path must never stomp on. `updateLane` filters
    // through PATCHABLE, which allows `status` and forbids `tool`.
    const next = active ? "running" : "idle";
    if (
      !lane.run_id &&
      (lane.status === "idle" || lane.status === "running") &&
      next !== lane.status
    ) {
      try {
        lanesLib.updateLane(lane.id, { status: next });
        changed = true;
      } catch {
        /* as above */
      }
    }
    if (changed) touched.add(lane.id);
  }

  for (const id of touched) {
    try {
      if (broadcast) broadcast(id);
    } catch {
      /* a broadcast failure must not abort the pass */
    }
  }
  return touched.size;
}

/**
 * Unref'd interval so it never blocks shutdown; disable with
 * DASHBOARD_CODEX_POLL_MS=0. Codex appends its journal continuously, so a 15s
 * beat keeps a codex card within one node of live without ever re-reading what
 * it has already ingested.
 */
function startCodexPoll(broadcast) {
  const POLL_MS = process.env.DASHBOARD_CODEX_POLL_MS
    ? Number(process.env.DASHBOARD_CODEX_POLL_MS)
    : 15_000;
  if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return;
  const timer = setInterval(() => {
    try {
      ingestCodexOnce({ broadcast });
    } catch {
      /* a poll pass never throws into the event loop */
    }
  }, POLL_MS);
  if (timer.unref) timer.unref();
}

module.exports = {
  eventsFromPayload,
  rolloutCwd,
  ingestCodexOnce,
  startCodexPoll,
  __resetOffsets,
};