/** * @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ĩ */ 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 }); }); });