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

/**
 * @file Tests server/lib/codex-usage.js — Codex sessions and token usage read
 * from rollout journals: the token mapping (input includes cached, output
 * includes reasoning), duplicate totals, a mid-session model switch, a fork's
 * copied history skipped, a sub-agent thread folded into its root session,
 * incremental reads, a deleted session not resurrected, and a Claude row never
 * touched. CODEX_HOME and the DB live in a temp dir.
 * @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 ROOT = fs.mkdtempSync(path.join(os.tmpdir(), "agenthub-codex-usage-"));
process.env.CLAUDE_HOME = path.join(ROOT, ".claude");
process.env.CODEX_HOME = path.join(ROOT, "codex");
process.env.DASHBOARD_DB_PATH = path.join(ROOT, "dashboard.db");

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

const { db, stmts } = require("../db");
const { parseRolloutChunk, ingestCodexUsage, settleActive } = require("../lib/codex-usage");

after(() => fs.rmSync(ROOT, { recursive: true, force: true }));

const DAY = path.join(process.env.CODEX_HOME, "sessions", "2026", "09", "29");

function meta(id, extra = {}) {
  return {
    timestamp: "2026-09-29T01:00:00.000Z",
    type: "session_meta",
    payload: { id, session_id: id, timestamp: "2026-09-29T01:00:00.000Z", cwd: "/w/p", ...extra },
  };
}
const ctx = (model) => ({
  timestamp: "2026-09-29T01:00:01.000Z",
  type: "turn_context",
  payload: { model },
});
function tc(total, last, ts = "2026-09-29T01:00:02.000Z") {
  const v = (t) => ({
    input_tokens: t[0],
    cached_input_tokens: t[1],
    cache_write_input_tokens: 0,
    output_tokens: t[2],
    reasoning_output_tokens: 0,
    total_tokens: t[0] + t[2],
  });
  return {
    timestamp: ts,
    type: "event_msg",
    payload: {
      type: "token_count",
      info: { total_token_usage: v(total), last_token_usage: v(last) },
    },
  };
}
const jsonl = (lines) => lines.map((l) => JSON.stringify(l)).join("\n") + "\n";
function writeRollout(id, lines, mtime = new Date("2026-09-29T02:00:00Z")) {
  fs.mkdirSync(DAY, { recursive: true });
  const file = path.join(DAY, `rollout-2026-09-29T01-00-00-${id}.jsonl`);
  fs.writeFileSync(file, jsonl(lines));
  fs.utimesSync(file, mtime, mtime);
  return file;
}
const usage = (sid) =>
  db
    .prepare(
      "SELECT model, input_tokens AS i, cache_read_tokens AS r, output_tokens AS o FROM token_usage WHERE session_id = ? ORDER BY model"
    )
    .all(sid);

beforeEach(() => {
  fs.rmSync(process.env.CODEX_HOME, { recursive: true, force: true });
  db.exec(
    "DELETE FROM codex_rollouts; DELETE FROM token_usage; DELETE FROM events; DELETE FROM sessions;"
  );
});

describe("parseRolloutChunk", () => {
  it("maps tokens, skips a repeated total and a null info, and follows the model", () => {
    const text = jsonl([
      meta("t1"),
      ctx("gpt-a"),
      tc([100, 60, 10], [100, 60, 10]),
      tc([100, 60, 10], [100, 60, 10]), // emitted twice
      { type: "event_msg", payload: { type: "token_count", info: null } },
      ctx("gpt-b"),
      tc([150, 90, 15], [50, 30, 5], "2026-09-29T01:05:00.000Z"),
    ]);
    const s = parseRolloutChunk(text, {});
    assert.equal(s.threadId, "t1");
    assert.equal(s.sessionId, "t1");
    assert.equal(s.cwd, "/w/p");
    assert.equal(s.model, "gpt-b");
    assert.deepEqual(s.usage, {
      "gpt-a": { input: 40, cacheRead: 60, cacheWrite: 0, output: 10 },
      "gpt-b": { input: 20, cacheRead: 30, cacheWrite: 0, output: 5 },
    });
  });

  it("continues from saved state across chunks", () => {
    let s = parseRolloutChunk(jsonl([meta("t1"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])]), {});
    s = parseRolloutChunk(jsonl([tc([10, 0, 1], [10, 0, 1]), tc([30, 0, 3], [20, 0, 2])]), s);
    assert.deepEqual(s.usage["gpt-a"], { input: 30, cacheRead: 0, cacheWrite: 0, output: 3 });
  });
});

describe("ingestCodexUsage", () => {
  it("creates a codex session with its usage, named from session_index.jsonl", async () => {
    writeRollout("s1", [meta("s1"), ctx("gpt-a"), tc([100, 60, 10], [100, 60, 10])]);
    fs.writeFileSync(
      path.join(process.env.CODEX_HOME, "session_index.jsonl"),
      JSON.stringify({ id: "s1", thread_name: "Fix the thing" }) + "\n"
    );
    const events = [];
    await ingestCodexUsage({
      broadcast: (t, d) => events.push([t, d.id]),
      now: Date.parse("2026-09-29T03:00:00Z"),
    });
    const row = stmts.getSession.get("s1");
    assert.equal(row.tool, "codex");
    assert.equal(row.name, "Fix the thing");
    assert.equal(row.cwd, "/w/p");
    assert.equal(row.model, "gpt-a");
    assert.equal(row.status, "completed");
    assert.equal(row.started_at, "2026-09-29T01:00:00.000Z");
    assert.equal(row.ended_at, "2026-09-29T02:00:00.000Z"); // the journal's mtime
    assert.equal(row.updated_at, "2026-09-29T02:00:00.000Z"); // last activity, not ingestion
    assert.equal(row.transcript_path, null);
    assert.deepEqual(usage("s1"), [{ model: "gpt-a", i: 40, r: 60, o: 10 }]);
    assert.deepEqual(events, [["session_created", "s1"]]);
  });

  it("names a session with no thread name by its first prompt", async () => {
    writeRollout("s2", [
      meta("s2"),
      {
        type: "event_msg",
        payload: { type: "user_message", message: "  Fix the login bug\nthen ship" },
      },
      { type: "event_msg", payload: { type: "user_message", message: "second prompt" } },
      ctx("gpt-a"),
      tc([1, 0, 1], [1, 0, 1]),
    ]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.equal(stmts.getSession.get("s2").name, "Fix the login bug");
  });

  it("reads a prompt from the item_completed UserMessage shape too", async () => {
    writeRollout("s3", [
      meta("s3"),
      {
        type: "event_msg",
        payload: {
          type: "item_completed",
          item: { type: "UserMessage", content: [{ type: "text", text: "Continue the task" }] },
        },
      },
    ]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.equal(stmts.getSession.get("s3").name, "Continue the task");
  });

  it("is active while a journal was written in the last two minutes", async () => {
    const now = Date.now();
    writeRollout(
      "s1",
      [meta("s1"), ctx("gpt-a"), tc([1, 0, 1], [1, 0, 1])],
      new Date(now - 10_000)
    );
    await ingestCodexUsage({ now });
    assert.equal(stmts.getSession.get("s1").status, "active");
    assert.equal(stmts.getSession.get("s1").ended_at, null);
  });

  it("settles an active session to completed once its journals go quiet", async () => {
    const now = Date.now();
    writeRollout(
      "s1",
      [meta("s1"), ctx("gpt-a"), tc([1, 0, 1], [1, 0, 1])],
      new Date(now - 10_000)
    );
    await ingestCodexUsage({ now });
    const events = [];
    settleActive((t, d) => events.push([t, d.status]), now + 5 * 60_000);
    assert.equal(stmts.getSession.get("s1").status, "completed");
    assert.equal(
      stmts.getSession.get("s1").ended_at,
      new Date(Math.floor(now - 10_000)).toISOString()
    );
    assert.deepEqual(events, [["session_updated", "completed"]]);
  });

  it("skips the history a fork copied from its parent", async () => {
    const parent = [
      meta("p1"),
      ctx("gpt-a"),
      tc([100, 0, 10], [100, 0, 10]),
      tc([300, 0, 30], [200, 0, 20]),
    ];
    writeRollout("p1", parent);
    writeRollout("f1", [
      meta("f1", { forked_from_id: "p1" }),
      meta("p1"),
      ...parent.slice(1),
      tc([350, 0, 35], [50, 0, 5]),
    ]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.deepEqual(usage("p1"), [{ model: "gpt-a", i: 300, r: 0, o: 30 }]);
    assert.deepEqual(usage("f1"), [{ model: "gpt-a", i: 50, r: 0, o: 5 }]);
  });

  it("folds a sub-agent thread into its root session", async () => {
    writeRollout("r1", [meta("r1"), ctx("gpt-a"), tc([100, 0, 10], [100, 0, 10])]);
    writeRollout("c1", [
      meta("c1", { session_id: "r1", thread_source: "subagent" }),
      ctx("gpt-b"),
      tc([40, 0, 4], [40, 0, 4]),
    ]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.equal(stmts.getSession.get("c1"), undefined);
    assert.deepEqual(usage("r1"), [
      { model: "gpt-a", i: 100, r: 0, o: 10 },
      { model: "gpt-b", i: 40, r: 0, o: 4 },
    ]);
  });

  it("reads only what was appended, and re-reads a file that shrank", async () => {
    const lines = [meta("s1"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])];
    const file = writeRollout("s1", lines);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    fs.appendFileSync(file, jsonl([tc([30, 0, 3], [20, 0, 2])]) + '{"partial":');
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.deepEqual(usage("s1"), [{ model: "gpt-a", i: 30, r: 0, o: 3 }]);

    fs.writeFileSync(file, jsonl(lines)); // rewritten shorter
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.deepEqual(usage("s1"), [{ model: "gpt-a", i: 10, r: 0, o: 1 }]);
  });

  it("does not bring back a deleted session while its journal is unchanged", async () => {
    writeRollout("s1", [meta("s1"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    db.prepare("DELETE FROM sessions WHERE id = 's1'").run();
    db.prepare("DELETE FROM token_usage WHERE session_id = 's1'").run();
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.equal(stmts.getSession.get("s1"), undefined);
  });

  it("never touches a Claude session with the same id", async () => {
    stmts.insertSession.run("s1", "mine", "active", "/c", "claude-x", null);
    writeRollout("s1", [meta("s1"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    const row = stmts.getSession.get("s1");
    assert.equal(row.tool, "claude");
    assert.equal(row.name, "mine");
    assert.deepEqual(usage("s1"), []);
  });

  it("keeps skipping a fork's copied history that arrives after the first read", async () => {
    const parent = [
      meta("p1"),
      ctx("gpt-a"),
      tc([100, 0, 10], [100, 0, 10]),
      tc([300, 0, 30], [200, 0, 20]),
    ];
    writeRollout("p1", parent);
    const fork = writeRollout("f1", [meta("f1", { forked_from_id: "p1" }), meta("p1"), parent[1]]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    fs.appendFileSync(fork, jsonl([...parent.slice(2), tc([350, 0, 35], [50, 0, 5])]));
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.deepEqual(usage("f1"), [{ model: "gpt-a", i: 50, r: 0, o: 5 }]);
  });

  it("counts a fork whole, and says so, when its parent's journal is gone", async () => {
    const warn = console.warn;
    const warned = [];
    console.warn = (m) => warned.push(String(m));
    try {
      writeRollout("f1", [
        meta("f1", { forked_from_id: "gone" }),
        ctx("gpt-a"),
        tc([100, 0, 10], [100, 0, 10]),
      ]);
      await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    } finally {
      console.warn = warn;
    }
    assert.deepEqual(usage("f1"), [{ model: "gpt-a", i: 100, r: 0, o: 10 }]);
    assert.ok(warned.some((m) => m.includes("fork parent gone")));
  });

  it("does not count a journal twice when CODEX_HOME moves", async () => {
    writeRollout("s1", [meta("s1"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    const home = process.env.CODEX_HOME;
    fs.renameSync(home, `${home}-moved`);
    process.env.CODEX_HOME = `${home}-moved`;
    try {
      await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    } finally {
      fs.renameSync(`${home}-moved`, home);
      process.env.CODEX_HOME = home;
    }
    assert.deepEqual(usage("s1"), [{ model: "gpt-a", i: 10, r: 0, o: 1 }]);
    assert.equal(db.prepare("SELECT COUNT(*) AS n FROM codex_rollouts").get().n, 1);
  });

  it("does not call a journal written in the future active", async () => {
    const now = Date.now();
    writeRollout(
      "s1",
      [meta("s1"), ctx("gpt-a"), tc([1, 0, 1], [1, 0, 1])],
      new Date(now + 60_000)
    );
    await ingestCodexUsage({ now });
    assert.equal(stmts.getSession.get("s1").status, "completed");
  });

  it("drops the usage of a model a re-read journal no longer has", async () => {
    const file = writeRollout("s1", [
      meta("s1"),
      ctx("gpt-a"),
      tc([10, 0, 1], [10, 0, 1]),
      ctx("gpt-b"),
      tc([30, 0, 3], [20, 0, 2]),
    ]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.equal(usage("s1").length, 2);
    fs.writeFileSync(file, jsonl([meta("s1"), ctx("gpt-a"), tc([5, 0, 1], [5, 0, 1])]));
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.deepEqual(usage("s1"), [{ model: "gpt-a", i: 5, r: 0, o: 1 }]);
  });

  it("takes cwd, start and model from the root thread even when a sub-agent is read first", async () => {
    writeRollout("c1", [
      meta("c1", { session_id: "r1", cwd: "/w/child", timestamp: "2026-09-29T01:30:00.000Z" }),
      ctx("gpt-b"),
      tc([40, 0, 4], [40, 0, 4]),
    ]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.equal(stmts.getSession.get("r1").cwd, "/w/child");
    writeRollout(
      "r1",
      [meta("r1"), ctx("gpt-a"), tc([100, 0, 10], [100, 0, 10])],
      new Date("2026-09-29T01:40:00Z")
    );
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    const row = stmts.getSession.get("r1");
    assert.equal(row.cwd, "/w/p");
    assert.equal(row.started_at, "2026-09-29T01:00:00.000Z");
    assert.equal(row.model, "gpt-a"); // the root's, though the child's file is newer
  });

  it("reads a journal with a line longer than one read in bounded steps", async () => {
    const file = writeRollout("s1", [meta("s1"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])]);
    fs.appendFileSync(file, `{"type":"response_item","x":"${"y".repeat(17 * 1024 * 1024)}"}\n`);
    fs.appendFileSync(file, jsonl([tc([30, 0, 3], [20, 0, 2])]));
    for (let i = 0; i < 4; i++) await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.deepEqual(usage("s1"), [{ model: "gpt-a", i: 30, r: 0, o: 3 }]);
    const row = db.prepare("SELECT offset, size FROM codex_rollouts").get();
    assert.equal(row.offset, row.size);
  });

  it("leaves a hook-fed session's status to its hooks, and still writes its usage", async () => {
    stmts.insertSession.run("s1", "Codex s1", "active", "/w/p", null, null);
    db.prepare("UPDATE sessions SET tool = 'codex' WHERE id = 's1'").run();
    stmts.insertEvent.run("s1", null, "Stop", null, "ready", "{}");
    const before = stmts.getSession.get("s1");
    writeRollout("s1", [meta("s1"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    settleActive(() => {}, Date.parse("2026-09-29T09:00:00Z"));
    const row = stmts.getSession.get("s1");
    assert.equal(row.status, "active");
    assert.equal(row.ended_at, null);
    assert.equal(row.updated_at, before.updated_at);
    assert.equal(row.model, "gpt-a");
    assert.deepEqual(usage("s1"), [{ model: "gpt-a", i: 10, r: 0, o: 1 }]);
  });

  it("hands the status back to the journal once the hooks stop reporting", async () => {
    // Hooked once (an hour before), then the hooks were removed; the journal
    // kept being written.
    stmts.insertSession.run("s1", "Codex s1", "abandoned", "/w/p", null, null);
    db.prepare("UPDATE sessions SET tool = 'codex' WHERE id = 's1'").run();
    db.prepare(
      "INSERT INTO events (session_id, event_type, created_at) VALUES ('s1', 'Stop', '2026-09-29T01:00:00.000Z')"
    ).run();
    const now = Date.parse("2026-09-29T02:00:30Z");
    writeRollout(
      "s1",
      [meta("s1"), ctx("gpt-a"), tc([30, 0, 3], [30, 0, 3])],
      new Date(now - 30_000)
    );
    await ingestCodexUsage({ now });
    assert.equal(stmts.getSession.get("s1").status, "active");
    settleActive(() => {}, now + 5 * 60_000);
    assert.equal(stmts.getSession.get("s1").status, "completed");
  });

  it("imports its journals again after Clear data", async () => {
    writeRollout("s1", [meta("s1"), ctx("gpt-a"), tc([5, 0, 1], [5, 0, 1])]);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    const express = require("express");
    const http = require("node:http");
    const app = express();
    app.use(express.json());
    app.use("/api/settings", require("../routes/settings"));
    const server = http.createServer(app);
    await new Promise((r) => server.listen(0, "127.0.0.1", r));
    try {
      await fetch(`http://127.0.0.1:${server.address().port}/api/settings/clear-data`, {
        method: "POST",
      });
    } finally {
      server.close();
    }
    assert.equal(stmts.getSession.get("s1"), undefined);
    await ingestCodexUsage({ now: Date.parse("2026-09-29T03:00:00Z") });
    assert.deepEqual(usage("s1"), [{ model: "gpt-a", i: 5, r: 0, o: 1 }]);
  });

  it("answers quietly when CODEX_HOME has no sessions", async () => {
    const r = await ingestCodexUsage({ now: Date.now() });
    assert.deepEqual(r, { files: 0, sessions: 0 });
  });
});