AgentHub/server/__tests__/codex-ingest.test.js

/**
 * @file Unit tests for the Codex lane status mirror: one pass over fixture
 * rollout journals flips a codex lane idle↔running, broadcasts it, and never
 * touches a lane's stage.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

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

const SUITE_DB_PATH = path.join(
  os.tmpdir(),
  `dashboard-codex-ingest-${Date.now()}-${process.pid}.db`
);
process.env.DASHBOARD_DB_PATH = SUITE_DB_PATH;

const { describe, it, before, after } = require("node:test");
const assert = require("node:assert/strict");

const codex = require("../lib/codex-ingest");
const lanes = require("../lib/lanes");

/** One rollout line in the real on-disk shape: top-level envelope + payload. */
function line(payload, type = "response_item") {
  return JSON.stringify({ timestamp: new Date().toISOString(), ordinal: 1, type, payload });
}

describe("codex status mirror", () => {
  const HOME = fs.mkdtempSync(path.join(os.tmpdir(), "codex-home-"));
  const DIR = path.join(HOME, "sessions", "2026", "08", "27");
  let prevHome;
  let seq = 0;

  before(() => {
    prevHome = process.env.CODEX_HOME;
    process.env.CODEX_HOME = HOME;
    fs.mkdirSync(DIR, { recursive: true });
  });

  after(() => {
    if (prevHome === undefined) delete process.env.CODEX_HOME;
    else process.env.CODEX_HOME = prevHome;
    codex.__resetCache();
    fs.rmSync(HOME, { recursive: true, force: true });
  });

  /** A rollout journal for `cwd` with one `npm run test:server` shell call. */
  function writeRollout(cwd, { mtime } = {}) {
    seq += 1;
    const file = path.join(DIR, `rollout-2026-08-27T10-00-00-${seq}.jsonl`);
    fs.writeFileSync(
      file,
      [
        line({ cwd, session_id: `s${seq}` }, "session_meta"),
        line({
          type: "function_call",
          name: "exec_command",
          arguments: JSON.stringify({ cmd: "npm run test:server", workdir: cwd }),
        }),
      ].join("\n") + "\n"
    );
    if (mtime !== undefined) fs.utimesSync(file, mtime / 1000, mtime / 1000);
    codex.__resetCache();
    return file;
  }

  const codexLane = (prefix = "codex-lane-") =>
    lanes.createLane({
      title: "cx",
      cwd: fs.mkdtempSync(path.join(os.tmpdir(), prefix)),
      tool: "codex",
    });

  it("marks a codex lane running while its journal is fresh, and broadcasts it", () => {
    const lane = codexLane();
    writeRollout(lane.cwd);

    const seen = [];
    assert.equal(codex.ingestCodexOnce({ broadcast: (id) => seen.push(id) }), 1);
    assert.equal(lanes.getLane(lane.id).status, "running");
    assert.deepEqual(seen, [lane.id]);
    lanes.deleteLane(lane.id);
  });

  it("never writes a stage or a detection, whatever the journal ran", () => {
    const lane = codexLane();
    writeRollout(lane.cwd);

    codex.ingestCodexOnce({});
    const after = lanes.getLane(lane.id);
    assert.equal(after.stage, "idle");
    assert.equal(after.detected_stage, null);
    assert.equal(after.detected_signal, null);
    lanes.deleteLane(lane.id);
  });

  it("keeps a lane running while an older journal of the same lane sits beside the live one", () => {
    const lane = codexLane();
    writeRollout(lane.cwd, { mtime: Date.now() - 6 * 60 * 60 * 1000 });
    writeRollout(lane.cwd);

    codex.ingestCodexOnce({});
    assert.equal(lanes.getLane(lane.id).status, "running");
    assert.equal(codex.ingestCodexOnce({}), 0);
    assert.equal(lanes.getLane(lane.id).status, "running");
    lanes.deleteLane(lane.id);
  });

  it("counts a journal whose mtime is a fraction of a second ahead as fresh", () => {
    const lane = codexLane();
    writeRollout(lane.cwd, { mtime: Date.now() + 400 });

    assert.equal(codex.ingestCodexOnce({}), 1);
    assert.equal(lanes.getLane(lane.id).status, "running");
    lanes.deleteLane(lane.id);
  });

  it("leaves a lane alone while its session's hooks are current", () => {
    const { stmts } = require("../db");
    const lane = codexLane();
    lanes.updateLane(lane.id, { session_id: "hooked" });
    stmts.insertSession.run("hooked", "x", "active", lane.cwd, null, null);
    stmts.insertEvent.run("hooked", null, "Stop", null, "ready", "{}");
    writeRollout(lane.cwd); // fresh journal: the mirror alone would say running

    assert.equal(codex.ingestCodexOnce({}), 0);
    assert.equal(lanes.getLane(lane.id).status, "idle");
    lanes.deleteLane(lane.id);
  });

  it("a second pass with nothing new changes nothing", () => {
    const lane = codexLane();
    writeRollout(lane.cwd);

    codex.ingestCodexOnce({});
    assert.equal(codex.ingestCodexOnce({}), 0);
    lanes.deleteLane(lane.id);
  });

  it("puts a running lane back to idle once its journal goes quiet", () => {
    const lane = codexLane();
    lanes.updateLane(lane.id, { status: "running" });
    writeRollout(lane.cwd, { mtime: Date.now() - 6 * 60 * 60 * 1000 });

    assert.equal(codex.ingestCodexOnce({}), 1);
    assert.equal(lanes.getLane(lane.id).status, "idle");
    lanes.deleteLane(lane.id);
  });

  it("treats a journal whose mtime is in the FUTURE as idle", () => {
    // A clock stepped back (or a restore that preserved timestamps) leaves a
    // finished thread's mtime ahead of now; raw `now - mtime` would read as live.
    const lane = codexLane();
    writeRollout(lane.cwd, { mtime: Date.now() + 6 * 60 * 60 * 1000 });

    assert.equal(codex.ingestCodexOnce({}), 0);
    assert.equal(lanes.getLane(lane.id).status, "idle");
    lanes.deleteLane(lane.id);
  });

  it("never touches a claude lane, even at the journal's own cwd", () => {
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "claude-lane-"));
    const lane = lanes.createLane({ title: "cc", cwd, tool: "claude" });
    writeRollout(cwd);

    assert.equal(codex.ingestCodexOnce({}), 0);
    assert.equal(lanes.getLane(lane.id).status, "idle");
    lanes.deleteLane(lane.id);
  });

  it("ignores a journal whose cwd belongs to no lane", () => {
    const lane = codexLane();
    writeRollout(fs.mkdtempSync(path.join(os.tmpdir(), "codex-orphan-")));

    assert.equal(codex.ingestCodexOnce({}), 0);
    assert.equal(lanes.getLane(lane.id).status, "idle");
    lanes.deleteLane(lane.id);
  });

  it("handles a journal whose session_meta is unparseable gracefully", () => {
    const lane = codexLane();
    fs.writeFileSync(path.join(DIR, "rollout-bad-meta.jsonl"), "not valid json at all\n");
    codex.__resetCache();

    assert.equal(codex.ingestCodexOnce({}), 0);
    assert.equal(lanes.getLane(lane.id).status, "idle");
    lanes.deleteLane(lane.id);
  });

  it("picks the deepest lane when a journal's cwd matches nested lanes", () => {
    const outer = fs.mkdtempSync(path.join(os.tmpdir(), "codex-outer-"));
    const inner = fs.mkdtempSync(path.join(outer, "inner"));
    const outerLane = lanes.createLane({ title: "outer", cwd: outer, tool: "codex" });
    const innerLane = lanes.createLane({ title: "inner", cwd: inner, tool: "codex" });
    writeRollout(inner);

    const seen = [];
    assert.equal(codex.ingestCodexOnce({ broadcast: (id) => seen.push(id) }), 1);
    assert.equal(lanes.getLane(innerLane.id).status, "running");
    assert.equal(lanes.getLane(outerLane.id).status, "idle");
    assert.deepEqual(seen, [innerLane.id]);
    lanes.deleteLane(innerLane.id);
    lanes.deleteLane(outerLane.id);
  });

  it("does not touch status of a lane with an active run_id", () => {
    const lane = codexLane();
    lanes.updateLane(lane.id, { run_id: "active-run" });
    writeRollout(lane.cwd);

    codex.ingestCodexOnce({});
    assert.equal(lanes.getLane(lane.id).status, "idle");
    lanes.deleteLane(lane.id);
  });

  it("does not change status from provisioning or failed", () => {
    const lane = codexLane();
    lanes.updateLane(lane.id, { status: "provisioning" });
    writeRollout(lane.cwd);

    codex.ingestCodexOnce({});
    assert.equal(lanes.getLane(lane.id).status, "provisioning");

    lanes.updateLane(lane.id, { status: "failed" });
    codex.ingestCodexOnce({});
    assert.equal(lanes.getLane(lane.id).status, "failed");
    lanes.deleteLane(lane.id);
  });

  it("respects DASHBOARD_CODEX_POLL_MS=0 to disable polling", () => {
    const oldMs = process.env.DASHBOARD_CODEX_POLL_MS;
    try {
      process.env.DASHBOARD_CODEX_POLL_MS = "0";
      assert.equal(
        codex.startCodexPoll(() => {}),
        undefined
      );
    } finally {
      if (oldMs === undefined) delete process.env.DASHBOARD_CODEX_POLL_MS;
      else process.env.DASHBOARD_CODEX_POLL_MS = oldMs;
    }
  });
});