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

/**
 * @file Unit tests for Codex stage ingestion: the payload→hook-event mapping,
 * and one full pass over a fixture rollout journal (detection + status mirror +
 * broadcast + the no-new-bytes no-op).
 * @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 payload mapping", () => {
  it("maps a function_call exec_command to a Bash event", () => {
    const events = codex.eventsFromPayload({
      type: "function_call",
      name: "exec_command",
      arguments: JSON.stringify({ cmd: "npm run test:server", workdir: "/repo" }),
    });
    assert.deepEqual(events, [
      { tool_name: "Bash", tool_input: { command: "npm run test:server" } },
    ]);
  });

  it("maps a custom_tool_call exec to a Bash event carrying the raw script", () => {
    const [event] = codex.eventsFromPayload({
      type: "custom_tool_call",
      name: "exec",
      input: 'const r = await tools.exec_command({cmd:"git push origin feat/x"}); text(r.output);',
    });
    assert.equal(event.tool_name, "Bash");
    assert.match(event.tool_input.command, /git push origin feat\/x/);
  });

  it("maps an apply_patch script to one event per touched file", () => {
    const events = codex.eventsFromPayload({
      type: "custom_tool_call",
      name: "exec",
      input:
        'const patch = "*** Begin Patch\n*** Update File: server/lib/lanes.js\n+x\n' +
        '*** Add File: docs/NEW.md\n+y\n*** End Patch";',
    });
    assert.deepEqual(events, [
      { tool_name: "Edit", tool_input: { file_path: "server/lib/lanes.js" } },
      { tool_name: "Write", tool_input: { file_path: "docs/NEW.md" } },
    ]);
  });

  it("maps an apply_patch script whose newlines are still escaped", () => {
    // Verbatim from a real rollout: the patch heredoc lives inside a JS string
    // literal, so after JSON.parse the line breaks are a literal backslash-n,
    // not newlines. Matching only real newlines silently sees no files at all.
    const events = codex.eventsFromPayload({
      type: "custom_tool_call",
      name: "exec",
      input:
        'const patch = "*** Begin Patch\\n*** Add File: config/demo.yml\\n+# demo\\n*** End Patch";',
    });
    assert.deepEqual(events, [
      { tool_name: "Write", tool_input: { file_path: "config/demo.yml" } },
    ]);
  });

  it("maps update_plan to TaskCreate and spawn_agent to Agent", () => {
    assert.deepEqual(
      codex.eventsFromPayload({ type: "function_call", name: "update_plan", arguments: "{}" }),
      [{ tool_name: "TaskCreate", tool_input: {} }]
    );
    assert.deepEqual(
      codex.eventsFromPayload({
        type: "function_call",
        name: "spawn_agent",
        arguments: JSON.stringify({ task_name: "code_reviewer", message: "…" }),
      }),
      [{ tool_name: "Agent", tool_input: { subagent_type: "code_reviewer" } }]
    );
  });

  it("ignores anything that is not a tool call", () => {
    assert.deepEqual(codex.eventsFromPayload({ type: "reasoning" }), []);
    assert.deepEqual(
      codex.eventsFromPayload({ type: "custom_tool_call_output", name: "exec" }),
      []
    );
    assert.deepEqual(codex.eventsFromPayload({ type: "function_call", name: "web_search" }), []);
    assert.deepEqual(codex.eventsFromPayload(null), []);
  });
});

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

  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.__resetOffsets();
    try {
      fs.rmSync(HOME, { recursive: true, force: true });
    } catch {
      /* best-effort */
    }
  });

  /** A rollout journal for `cwd` with one `npm run test:server` shell call. */
  function writeRollout(name, cwd) {
    const file = path.join(DIR, name);
    fs.writeFileSync(
      file,
      [
        line({ cwd, session_id: "abc" }, "session_meta"),
        line({
          type: "function_call",
          name: "exec_command",
          arguments: JSON.stringify({ cmd: "npm run test:server", workdir: cwd }),
        }),
      ].join("\n") + "\n"
    );
    return file;
  }

  it("detects a stage for a codex lane and mirrors its status", () => {
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-lane-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    writeRollout("rollout-2026-08-27T10-00-00-aaa.jsonl", cwd);
    codex.__resetOffsets();

    const seen = [];
    assert.equal(codex.ingestCodexOnce({ broadcast: (id) => seen.push(id) }), 1);
    const after = lanes.getLane(lane.id);
    assert.equal(after.detected_stage, "tests"); // `npm run test:server` → default template's tests node
    assert.match(after.detected_signal, /test:server/);
    assert.equal(after.status, "running"); // a journal written just now means work is happening
    assert.deepEqual(seen, [lane.id]);
    lanes.deleteLane(lane.id);
  });

  it("a second pass with no appended bytes changes nothing", () => {
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-lane-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    writeRollout("rollout-2026-08-27T10-00-01-bbb.jsonl", cwd);
    codex.__resetOffsets();

    codex.ingestCodexOnce({});
    const first = lanes.getLane(lane.id).detected_at;
    assert.equal(codex.ingestCodexOnce({}), 0);
    assert.equal(lanes.getLane(lane.id).detected_at, first);
    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("rollout-2026-08-27T10-00-02-ccc.jsonl", cwd);
    codex.__resetOffsets();

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

  it("ignores a journal whose cwd belongs to no lane", () => {
    const orphan = fs.mkdtempSync(path.join(os.tmpdir(), "codex-orphan-"));
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-lane-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    writeRollout("rollout-2026-08-27T10-00-03-ddd.jsonl", orphan);
    codex.__resetOffsets();

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

  it("keeps byte offsets exact when a tail read starts mid multi-byte character", () => {
    // The first sight of a >64 KiB journal reads only its tail, and that offset
    // can land inside a 3-byte UTF-8 sequence. Splitting the decoded STRING and
    // measuring it back with Buffer.byteLength then counts the decoder's U+FFFD
    // replacement instead of the bytes actually consumed, so the next pass
    // starts 1-2 bytes off and every later record is unparseable.
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-lane-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    const file = path.join(DIR, "rollout-2026-08-27T10-00-06-utf8.jsonl");
    const record = (cmd) =>
      line({
        type: "function_call",
        name: "exec_command",
        arguments: JSON.stringify({ cmd, workdir: cwd }),
      });

    // Grow an all-multi-byte filler line until the tail boundary (size - 64 KiB)
    // falls on a UTF-8 continuation byte.
    const head = line({ cwd, session_id: "utf8" }, "session_meta") + "\n";
    const tailRecord = record("npm run test:server") + "\n";
    let filler = "→".repeat(30_000);
    for (let pad = 0; pad < 8; pad += 1) {
      const body = head + filler + "\n" + tailRecord;
      const buf = Buffer.from(body, "utf8");
      const boundary = buf.length - 64 * 1024;
      if (boundary > 0 && (buf[boundary] & 0xc0) === 0x80) break;
      filler += "x";
    }
    fs.writeFileSync(file, head + filler + "\n" + tailRecord);
    const buf = Buffer.from(fs.readFileSync(file));
    assert.equal(buf[buf.length - 64 * 1024] & 0xc0, 0x80, "fixture must straddle a UTF-8 char");

    codex.__resetOffsets();
    assert.equal(codex.ingestCodexOnce({}), 1);
    assert.equal(lanes.getLane(lane.id).detected_stage, "tests");

    // The real proof: the next append must still parse, which it only does if
    // the stored offset is exactly on the previous line boundary.
    fs.appendFileSync(file, record("git push origin feat/x") + "\n");
    assert.equal(codex.ingestCodexOnce({}), 1);
    assert.equal(lanes.getLane(lane.id).detected_stage, "ship");
    lanes.deleteLane(lane.id);
  });

  it("ingests a line appended exactly at the previous read's boundary", () => {
    // The offset after a pass lands on a line boundary, so the next chunk's
    // FIRST line is a whole record — dropping it as torn loses every append
    // that arrives one line at a time, which is how a journal actually grows.
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-lane-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    const file = path.join(DIR, "rollout-2026-08-27T10-00-04-fff.jsonl");
    fs.writeFileSync(file, line({ cwd, session_id: "boundary" }, "session_meta") + "\n");
    codex.__resetOffsets();
    codex.ingestCodexOnce({}); // consumes the header, offset now at EOF
    assert.equal(lanes.getLane(lane.id).detected_stage, null);

    fs.appendFileSync(
      file,
      line({
        type: "function_call",
        name: "exec_command",
        arguments: JSON.stringify({ cmd: "npm run test:server", workdir: cwd }),
      }) + "\n"
    );
    assert.equal(codex.ingestCodexOnce({}), 1);
    assert.equal(lanes.getLane(lane.id).detected_stage, "tests");
    lanes.deleteLane(lane.id);
  });

  it("re-reads a torn line whole on the next pass", () => {
    // A pass that catches the writer mid-line must not consume that fragment:
    // the record it belongs to would be lost forever.
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-lane-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    const file = path.join(DIR, "rollout-2026-08-27T10-00-05-ggg.jsonl");
    const record = line({
      type: "function_call",
      name: "exec_command",
      arguments: JSON.stringify({ cmd: "npm run test:server", workdir: cwd }),
    });
    fs.writeFileSync(
      file,
      line({ cwd, session_id: "torn" }, "session_meta") + "\n" + record.slice(0, 40)
    );
    codex.__resetOffsets();
    codex.ingestCodexOnce({});
    assert.equal(lanes.getLane(lane.id).detected_stage, null);

    fs.appendFileSync(file, record.slice(40) + "\n"); // the writer finishes the line
    assert.equal(codex.ingestCodexOnce({}), 1);
    assert.equal(lanes.getLane(lane.id).detected_stage, "tests");
    lanes.deleteLane(lane.id);
  });

  it("does not replay a journal it sees for the first time already idle", () => {
    // A thread that finished days ago must not paint a stage now — the first
    // sight of a stale file only records where it ends.
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-lane-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    const file = writeRollout("rollout-2026-08-20T10-00-00-eee.jsonl", cwd);
    const old = Date.now() - 6 * 60 * 60 * 1000;
    fs.utimesSync(file, old / 1000, old / 1000);
    codex.__resetOffsets();

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

  it("ignores a torn/partial JSONL line at the end of a file", () => {
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-torn-line-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    const file = path.join(DIR, "rollout-torn.jsonl");
    fs.writeFileSync(
      file,
      [
        line({ cwd, session_id: "torn" }, "session_meta"),
        line({
          type: "function_call",
          name: "exec_command",
          arguments: JSON.stringify({ cmd: "npm run test:server", workdir: cwd }),
        }),
        // A torn last line: not valid JSON, so it will be skipped on next pass
        '{"timestamp":"2026-08-27T10:00:00.000Z","type":"response_item","paylo',
      ].join("\n")
    );
    codex.__resetOffsets();

    const seen = [];
    assert.equal(codex.ingestCodexOnce({ broadcast: (id) => seen.push(id) }), 1);
    const after = lanes.getLane(lane.id);
    assert.equal(after.detected_stage, "tests"); // the valid line was processed
    assert.deepEqual(seen, [lane.id]);
    lanes.deleteLane(lane.id);
  });

  it("handles a journal whose session_meta is unparseable gracefully", () => {
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-bad-meta-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    const file = path.join(DIR, "rollout-bad-meta.jsonl");
    // Write an invalid JSON line as the first line
    fs.writeFileSync(file, "not valid json at all\n");
    codex.__resetOffsets();

    assert.equal(codex.ingestCodexOnce({}), 0);
    assert.equal(lanes.getLane(lane.id).detected_stage, null);
    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" });
    const file = path.join(DIR, "rollout-nested.jsonl");
    fs.writeFileSync(
      file,
      [
        line({ cwd: inner, session_id: "nested" }, "session_meta"),
        line({
          type: "function_call",
          name: "exec_command",
          arguments: JSON.stringify({ cmd: "npm run test:server", workdir: inner }),
        }),
      ].join("\n") + "\n"
    );
    codex.__resetOffsets();

    const seen = [];
    assert.equal(codex.ingestCodexOnce({ broadcast: (id) => seen.push(id) }), 1);
    // Inner lane should be the one detected, not outer
    assert.equal(lanes.getLane(innerLane.id).detected_stage, "tests");
    assert.equal(lanes.getLane(outerLane.id).detected_stage, null);
    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 cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-run-id-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    lanes.updateLane(lane.id, { run_id: "active-run" });
    writeRollout("rollout-2026-08-27T10-00-10-rrr.jsonl", cwd);
    codex.__resetOffsets();

    codex.ingestCodexOnce({});
    const after = lanes.getLane(lane.id);
    // Status should not change because run_id is present
    assert.equal(after.status, "idle");
    lanes.deleteLane(lane.id);
  });

  it("does not change status from provisioning or failed", () => {
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-status-protected-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    lanes.updateLane(lane.id, { status: "provisioning" });
    writeRollout("rollout-2026-08-27T10-00-11-ppp.jsonl", cwd);
    codex.__resetOffsets();

    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";
      // startCodexPoll should return early and not set up an interval
      const timer = codex.startCodexPoll(() => {});
      assert.equal(timer, undefined);
    } finally {
      if (oldMs === undefined) delete process.env.DASHBOARD_CODEX_POLL_MS;
      else process.env.DASHBOARD_CODEX_POLL_MS = oldMs;
    }
  });

  it("handles a shell call with command as an array in arguments", () => {
    const events = codex.eventsFromPayload({
      type: "function_call",
      name: "exec_command",
      arguments: JSON.stringify({ command: ["npm", "run", "test:server"], workdir: "/repo" }),
    });
    // Should join array elements
    assert.deepEqual(events, [
      { tool_name: "Bash", tool_input: { command: "npm run test:server" } },
    ]);
  });

  it("ignores a payload with a non-string name", () => {
    // A payload with name as null or number should not match any tool
    const events1 = codex.eventsFromPayload({
      type: "function_call",
      name: null,
      arguments: "{}",
    });
    assert.deepEqual(events1, []);

    const events2 = codex.eventsFromPayload({
      type: "custom_tool_call",
      name: 123,
      input: "some script",
    });
    assert.deepEqual(events2, []);
  });

  it("uses the LAST detection in a batch to move the lane forward", () => {
    // Within one ingest pass, if multiple tool calls map to different stages,
    // recordDetection is called in order. Only the LAST one that actually writes
    // should move the lane forward (forward-only). Simulate two tool calls:
    // first impl call (implement), then tests call (tests).
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-batch-last-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });
    const file = path.join(DIR, "rollout-batch-last.jsonl");
    fs.writeFileSync(
      file,
      [
        line({ cwd, session_id: "batch" }, "session_meta"),
        // Edit event (implement node)
        line({
          type: "custom_tool_call",
          name: "exec",
          input: 'await tools.edit("src/main.js", "add x");',
        }),
        // Bash event (tests node)
        line({
          type: "function_call",
          name: "exec_command",
          arguments: JSON.stringify({ cmd: "npm run test:server", workdir: cwd }),
        }),
      ].join("\n") + "\n"
    );
    codex.__resetOffsets();

    const seen = [];
    assert.equal(codex.ingestCodexOnce({ broadcast: (id) => seen.push(id) }), 1);
    const after = lanes.getLane(lane.id);
    // Should detect the tests node (later in batch), not implement
    assert.equal(after.detected_stage, "tests");
    lanes.deleteLane(lane.id);
  });

  it("limits file scan to MAX_FILES newest files", () => {
    // Create more than MAX_FILES journal files and verify only newest are scanned
    const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "codex-max-files-"));
    const lane = lanes.createLane({ title: "cx", cwd, tool: "codex" });

    // Create 50 old files (more than MAX_FILES=40)
    const oldTime = Date.now() - 2 * 60 * 60 * 1000;
    for (let i = 0; i < 50; i++) {
      const file = path.join(DIR, `rollout-old-${String(i).padStart(2, "0")}.jsonl`);
      fs.writeFileSync(file, line({ cwd, session_id: `old-${i}` }, "session_meta") + "\n");
      fs.utimesSync(file, oldTime / 1000, oldTime / 1000);
    }
    codex.__resetOffsets();

    // The scan should not stall and should return (no detection expected since all files are stale)
    const result = codex.ingestCodexOnce({});
    // All files are stale and not active, so no broadcast
    assert.equal(result, 0);
    lanes.deleteLane(lane.id);
  });

  it("detects correctly with Delete File in patch heredoc", () => {
    const events = codex.eventsFromPayload({
      type: "custom_tool_call",
      name: "exec",
      input: 'const patch = "*** Begin Patch\\n*** Delete File: old/removed.js\\n*** End Patch";',
    });
    // Delete creates an Edit event (not Write, not ignored)
    assert.deepEqual(events, [{ tool_name: "Edit", tool_input: { file_path: "old/removed.js" } }]);
  });

  it("handles file paths with whitespace in patch patches", () => {
    const events = codex.eventsFromPayload({
      type: "custom_tool_call",
      name: "exec",
      input:
        'const patch = "*** Begin Patch\\n*** Update File:  src/my file.js  \\n*** End Patch";',
    });
    // File path should be trimmed
    assert.deepEqual(events, [{ tool_name: "Edit", tool_input: { file_path: "src/my file.js" } }]);
  });
});