AgentHub/server/lib/codex-ingest.js

/**
 * @file Codex lane status mirror. Without AgentHub's Codex hooks installed
 * (codex-hooks.js, opt-in and trusted by the user in Codex) nothing else tells
 * the dashboard that a `codex` lane's agent is working; while a lane's hooks
 * are current (an event for its session in the last 2 minutes) this mirror
 * leaves the lane to them. Its on-disk signal is the rollout journal Codex
 * 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":…}}
 *
 * This module maps each journal to the `codex` lane owning its cwd and mirrors
 * idle↔running from how recently the journal was written. It reads nothing but
 * the first line (for the cwd) and never touches a lane's stage — codex lanes
 * carry no pipeline. Fail-safe like workflow-ingest.js: a malformed journal 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. `allRolloutFiles()` is the one
 * function to re-point if that ever stops being true (codex-usage.js walks
 * the same files for sessions and token usage).
 * @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");
const { getCodexHome } = require("./codex-home");

const FUTURE_SLACK_MS = 1000;
const CODEX_ACTIVE_MS = 120_000; // a journal written within 2 min means the agent is working
const MAX_FILES = 40; // newest-first cap so a years-deep sessions/ tree can't stall a pass
const MAX_DEPTH = 4; // sessions/<YYYY>/<MM>/<DD>/<file>

/**
 * Every rollout journal under `<home>/sessions/` ({path, mtime, size}),
 * unsorted. `home` is CODEX_HOME unless a remote source's mirror names its own.
 */
function allRolloutFiles(home = getCodexHome()) {
  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(home, "sessions"), 0);
  return out;
}

/** The newest MAX_FILES rollout journals, newest mtime first. */
function rolloutFiles() {
  return allRolloutFiles()
    .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;
}

/** Test seam: forget every cached cwd. */
function __resetCache() {
  cwdCache.clear();
}

// Keyed by file path, so in a process that runs for months it 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 entries.
const MAX_TRACKED = MAX_FILES * 4;

function pruneTracking(seen) {
  if (cwdCache.size <= MAX_TRACKED) return;
  for (const key of cwdCache.keys()) if (!seen.has(key)) cwdCache.delete(key);
}

/**
 * One poll pass: mirror each codex lane's idle↔running from its journals.
 *
 * @param {{broadcast?: (laneId: number) => void}} [options]
 * @returns {number} How many lanes actually changed (and were broadcast).
 */
function ingestCodexOnce({ broadcast } = {}) {
  const lanesLib = require("./lanes");

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

  const touched = new Set();
  // One verdict per lane — running when ANY of its journals is fresh. Judging
  // file by file against the same status snapshot flipped a lane with an old
  // journal beside the live one between running and idle on every pass.
  const activeByLane = new Map();
  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;

    // 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". A file that claims to be from the future is
    // not evidence of an agent working now, so treat it as idle. The second of
    // slack is for the file's nanosecond mtime against Date.now()'s truncated
    // millisecond: a journal written this very millisecond reads as future.
    const age = Date.now() - file.mtime;
    const active = age > -FUTURE_SLACK_MS && age < CODEX_ACTIVE_MS;
    const seen = activeByLane.get(lane.id);
    activeByLane.set(lane.id, { lane, active: active || (seen ? seen.active : false) });
  }

  // Codex's hooks (codex-hooks.js) report a turn's start and end as they
  // happen; while they are current, a journal written a minute ago would call
  // a lane that just stopped "running".
  const { db } = require("../db");
  const hookedSince = new Date(Date.now() - CODEX_ACTIVE_MS).toISOString();
  const recentHook = db.prepare(
    "SELECT 1 FROM events WHERE session_id = ? AND created_at > ? LIMIT 1"
  );

  for (const { lane, active } of activeByLane.values()) {
    if (lane.session_id && recentHook.get(lane.session_id, hookedSince)) continue;
    // 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 });
        touched.add(lane.id);
      } catch {
        /* lane deleted mid-pass, or sqlite busy — a status mirror is never worth throwing for */
      }
    }
  }

  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's running/idle within one beat of live.
 */
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 = {
  allRolloutFiles,
  rolloutCwd,
  FUTURE_SLACK_MS,
  ingestCodexOnce,
  startCodexPoll,
  __resetCache,
};