From f031ce0835f1d4fa8778c07cd5955c5c3b7c84b4 Mon Sep 17 00:00:00 2001 From: "itarun.p" Date: Sun, 26 Jul 2026 05:27:28 +0700 Subject: [PATCH] feat(queue): reclaim the 84% of the queue nothing ever reclaimed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses #103. `sync --compact`, and the column invariant that lived only in prose now runs in `doctor`. Measured on a real install: 34,492 lines, 5,595 unique keys, 83.8% superseded, 11 MB — and readQueueData re-reads and re-dedups ALL of it on every endpoint call, 6-8 per dashboard refresh, auto-refresh every 30s. The design risk is near zero because THE READERS ALREADY DEFINE THE OUTPUT. readQueueData keeps the last row per `source|model|hour_start`, so compaction keeps the last RAW LINE per that key — not a re-serialised row. Byte-identical API responses follow by construction, not by luck: the surviving bytes are the exact bytes the reader would have kept. There is a test that reproduces the reader's dedup and asserts its output is identical before and after. Unparseable lines are KEPT. They are invisible to every reader already, so dropping them changes no response — but a partial write worth investigating should survive a routine maintenance command. THREE THINGS FOUND WHILE BUILDING THIS. 1. The invariant as written is WRONG, and my check inherited the error. Running it against a real 34,922-row queue reported 8,236 violations — every one source=codex, and in every case the difference was exactly reasoning_output_tokens. The rows were right: codex and every-code fold reasoning INTO output_tokens, and computeRowCost (pricing/index.js:309) already made that distinction by charging their reasoning at zero. CLAUDE.md said "sum of all columns" with no exception. Corrected there, and expectedTotal() is now the single rule shared by doctor and the #104 parser harness, which had copied the same too-strict version. After the fix: 13 findings, all in 2 real codex buckets whose total exceeds its columns by 5,337. Left as a WARN — genuine data, not a check artefact. 2. AUTOMATIC compaction is NOT shipped, though #103 proposed it. Wired up first, it immediately rewrote an 11 MB production queue (34,924 -> 5,636 lines) during `npm test`. Sync now RECOMMENDS `--compact` past the threshold and does not act. Rewriting the user's data file unasked turns every path that reaches sync into a data-mutating path. 3. The reason it could: test/init-uninstall.test.js spawned the notify handler in a CHILD PROCESS with `env: {...process.env}`, so overriding process.env.HOME in the parent never reached it and the child resolved ~/.tokentracker. Pre-existing — it had been appending rows to the maintainer's real queue all along; compaction only made it visible. The helper now requires an isolated `home` and throws without one, with a test for that. Verified: ci:local no longer changes the real queue's mtime. Interrupt safety is write-then-rename, so the file is either entirely the old one or entirely the new one; a test forces the rename to fail and asserts the original is intact and no temp file is left behind. ci:local exit 0: 914 root tests, 256 dashboard. --- CLAUDE.md | 9 +- src/commands/doctor.js | 2 + src/commands/sync.js | 67 ++++++++++ src/lib/doctor.js | 72 ++++++++++ src/lib/queue-compact.js | 220 ++++++++++++++++++++++++++++++ test/init-uninstall.test.js | 26 +++- test/parser-conformance.test.js | 11 +- test/queue-compact.test.js | 229 ++++++++++++++++++++++++++++++++ 8 files changed, 630 insertions(+), 6 deletions(-) create mode 100644 src/lib/queue-compact.js create mode 100644 test/queue-compact.test.js diff --git a/CLAUDE.md b/CLAUDE.md index 3cd785f1..ce1ce3eb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -57,7 +57,14 @@ input_tokens = non-cached input only (no cache reads/writes) cached_input_tokens = cache reads cache_creation_input_tokens = cache writes reasoning_output_tokens = reasoning tokens (Codex/every-code fold them into output_tokens for cost) -total_tokens = input + output + cache_creation + cache_read + reasoning_output (sum of all columns; Gemini-style rows that omit reasoning still pass invariants if you set the column to 0) +total_tokens = input + output + cache_creation + cache_read + reasoning_output + EXCEPT codex / every-code, where reasoning_output is ALREADY + counted inside output_tokens, so their total_tokens correctly + omits it. `computeRowCost` (pricing/index.js:309) makes the same + distinction and charges their reasoning at zero. Gemini-style rows + that omit reasoning pass with the column set to 0. + Enforced by `doctor` (queue.row_invariant) via + expectedTotal() in src/lib/queue-compact.js. ``` **Cost is computed from `input_tokens + output_tokens + cached_input_tokens + cache_creation_input_tokens + reasoning_output_tokens` only — never `total_tokens`** (`computeRowCost` in `src/lib/pricing/index.js`). If a new provider only fills `total_tokens` with input=0/output=0, the dashboard renders **$0 cost** regardless of pricing entries. Distribute the total across columns or extend `computeRowCost`. diff --git a/src/commands/doctor.js b/src/commands/doctor.js index 900ff75c..ee583304 100644 --- a/src/commands/doctor.js +++ b/src/commands/doctor.js @@ -12,6 +12,7 @@ async function cmdDoctor(argv = []) { const home = os.homedir(); const { trackerDir } = await resolveTrackerPaths({ home }); const configPath = path.join(trackerDir, "config.json"); + const queuePath = path.join(trackerDir, "queue.jsonl"); const configStatus = await readJsonStrict(configPath); const config = @@ -36,6 +37,7 @@ async function cmdDoctor(argv = []) { trackerDir, configPath, cliPath, + queuePath, }, }); diff --git a/src/commands/sync.js b/src/commands/sync.js index 86709304..f16bc1c0 100644 --- a/src/commands/sync.js +++ b/src/commands/sync.js @@ -6,6 +6,7 @@ const cp = require("node:child_process"); const readline = require("node:readline"); const { ensureDir, readJson, writeJson, openLock } = require("../lib/fs"); +const { analyzeQueue, compactQueue } = require("../lib/queue-compact"); const { listRolloutFiles, listClaudeProjectFiles, @@ -143,6 +144,13 @@ async function cmdSync(argv) { const cursorsPath = path.join(trackerDir, "cursors.json"); const queuePath = path.join(trackerDir, "queue.jsonl"); const queueStatePath = path.join(trackerDir, "queue.state.json"); + + // Maintenance path: compact and stop. Deliberately inside the lock, since + // the whole point is that a concurrent append must not land between the + // read and the rename. + if (opts.compact) { + return runCompaction(queuePath, { auto: opts.auto }); + } const projectQueuePath = path.join(trackerDir, "project.queue.jsonl"); const projectQueueStatePath = path.join(trackerDir, "project.queue.state.json"); const grokSignalPath = path.join(trackerDir, "grok-last-session.json"); @@ -982,6 +990,29 @@ async function cmdSync(argv) { totalBuckets, }; + // Nothing ever reclaimed superseded rows, so the file grows monotonically + // and every endpoint call re-parses all of it. #103 proposed compacting + // automatically past a threshold. This RECOMMENDS instead, deliberately. + // + // The reason is not caution in the abstract. Wiring the automatic version up + // first, it immediately rewrote an 11 MB production queue — 34,924 lines + // down to 5,636 — during `npm test`, because test/init-uninstall.test.js + // spawned the notify handler in a child process without an isolated HOME. + // That leak is fixed, but the lesson stands: rewriting the user's data file + // without being asked turns any path that reaches sync into a data-mutating + // path. `--compact` is one command and the user chooses to run it. + const queueHealth = analyzeQueue(queuePath); + if ( + !opts.auto && + queueHealth.parseable >= COMPACT_MIN_LINES && + queueHealth.ratio >= COMPACT_RATIO + ) { + process.stdout.write( + `\nQueue: ${Math.round(queueHealth.ratio * 100)}% of ${queueHealth.parseable} rows are superseded` + + ` (${Math.round(queueHealth.bytes / 1024)} KiB). Run \`tracker sync --compact\` to reclaim them.\n`, + ); + } + if (!opts.auto) { process.stdout.write( [ @@ -1002,6 +1033,40 @@ async function cmdSync(argv) { } } +// When a queue is mostly dead weight and big enough to be worth rewriting, sync +// says so. Below either threshold the rewrite costs more than it saves. +const COMPACT_RATIO = 0.5; +const COMPACT_MIN_LINES = 5000; + +// Reporting is not optional. The README tells users they can `cat` this file; +// silently rewriting it underneath them is the wrong default. +function runCompaction(queuePath, { auto = false } = {}) { + const result = compactQueue(queuePath); + if (auto) return result; + if (!result.changed) { + process.stdout.write(`Queue compaction: nothing to reclaim (${result.reason}).\n`); + return result; + } + const savedKiB = Math.round((result.bytesBefore - result.bytesAfter) / 1024); + process.stdout.write( + [ + "Queue compacted:", + `- Lines: ${result.totalLines} -> ${result.keptLines}`, + `- Superseded rows reclaimed: ${result.superseded}`, + `- Size: ${savedKiB} KiB smaller`, + result.malformed > 0 + ? `- Kept ${result.malformed} unparseable line(s) untouched — they are invisible to every reader, but they are still your bytes` + : null, + "", + ] + // Not filter(Boolean): that drops the trailing "" too, and the report ran + // straight into the next line of output. + .filter((line) => line !== null) + .join("\n"), + ); + return result; +} + function parseArgs(argv) { const out = { auto: false, @@ -1010,6 +1075,7 @@ function parseArgs(argv) { fromOpenclaw: false, drain: false, repairGrok: false, + compact: false, }; for (let i = 0; i < argv.length; i++) { const a = argv[i]; @@ -1019,6 +1085,7 @@ function parseArgs(argv) { else if (a === "--from-openclaw") out.fromOpenclaw = true; else if (a === "--drain") out.drain = true; else if (a === "--repair-grok") out.repairGrok = true; + else if (a === "--compact") out.compact = true; else throw new Error(`Unknown option: ${a}`); } return out; diff --git a/src/lib/doctor.js b/src/lib/doctor.js index 404ca5af..df591d5d 100644 --- a/src/lib/doctor.js +++ b/src/lib/doctor.js @@ -1,4 +1,5 @@ const fs = require("node:fs/promises"); +const { findRowViolations } = require("./queue-compact"); const { constants } = require("node:fs"); const path = require("node:path"); @@ -28,6 +29,9 @@ async function buildDoctorReport({ if (paths.cliPath) { checks.push(await checkCliEntrypoint(paths.cliPath)); } + if (paths.queuePath) { + checks.push(await checkQueueRows(paths.queuePath)); + } // No cloud reachability check: TokenTracker is local-only, so there is no // remote endpoint whose availability could affect anything here. @@ -358,8 +362,76 @@ function summarizeChecks(checks = []) { return summary; } +// CLAUDE.md states the column invariant in prose: +// +// total = input + output + cache_creation + cache_read + reasoning +// +// Nothing enforced it at runtime, so a miswritten or corrupt row was aggregated +// and rendered rather than flagged — and a parser bug of exactly that shape is +// the class CLAUDE.md records at 1.6-7x magnitude. Same conversion as the +// curated-expiry and version-lockstep checks: a rule that lived in a document +// starts running. +// +// A warn rather than a fail: the rows are already on disk and already being +// rendered, so failing the whole health check would be reporting a crisis the +// user cannot act on in the moment. What they can act on is knowing which rows, +// and how many. +const QUEUE_VIOLATIONS_SHOWN = 5; + +function queueCheck(status, detail, meta) { + return { id: "queue.row_invariant", status, detail, critical: false, meta }; +} + +async function readQueueRowsForDoctor(queuePath) { + const raw = await fs.readFile(queuePath, "utf8"); + const rows = []; + let malformed = 0; + for (const line of raw.split("\n")) { + if (!line.trim()) continue; + try { + rows.push(JSON.parse(line)); + } catch { + malformed += 1; + } + } + return { rows, malformed }; +} + +async function checkQueueRows(queuePath) { + let rows; + let malformed; + try { + ({ rows, malformed } = await readQueueRowsForDoctor(queuePath)); + } catch (err) { + if (err && err.code === "ENOENT") { + return queueCheck("ok", "no queue yet", { path: queuePath }); + } + return queueCheck("warn", `queue unreadable: ${err?.message || err}`, { path: queuePath }); + } + + const violations = findRowViolations(rows); + if (violations.length === 0 && malformed === 0) { + return queueCheck("ok", `${rows.length} rows satisfy the column invariant`, { + path: queuePath, + rows: rows.length, + }); + } + + const parts = []; + if (violations.length > 0) parts.push(`${violations.length} row problem(s)`); + if (malformed > 0) parts.push(`${malformed} unparseable line(s)`); + return queueCheck("warn", `${parts.join(", ")} in ${rows.length + malformed} line(s)`, { + path: queuePath, + rows: rows.length, + malformed, + violations: violations.length, + examples: violations.slice(0, QUEUE_VIOLATIONS_SHOWN), + }); +} + module.exports = { buildDoctorReport, + checkQueueRows, buildBrowserOpenerCheck, buildNodeVersionCheck, isHeadlessEnvironment, diff --git a/src/lib/queue-compact.js b/src/lib/queue-compact.js new file mode 100644 index 00000000..26861016 --- /dev/null +++ b/src/lib/queue-compact.js @@ -0,0 +1,220 @@ +"use strict"; + +// Compaction for the append-only queue, and the row invariant that lived in +// prose. +// +// Measured on a real install: 34,492 lines, 5,595 unique keys, 28,897 +// superseded (83.8%), 11 MB. Five sixths of the file is dead weight, and +// `readQueueData` in local-api.js re-reads and re-dedups ALL of it on every +// endpoint call — a dashboard refresh hits 6-8 endpoints, auto-refresh defaults +// to 30s. Nothing ever reclaims it; the only rewrite in the codebase is a +// one-off migration. +// +// The design risk here is close to zero for one reason: THE READERS ALREADY +// DEFINE THE OUTPUT. `readQueueData` keeps the last row per +// `source|model|hour_start`. Compaction only has to produce what every reader +// already computes, so it keeps the last RAW LINE per key — not a re-serialised +// row. Byte-identical API responses then follow by construction rather than by +// luck, because the surviving bytes are the exact bytes the reader would have +// kept. + +const fs = require("node:fs"); +const path = require("node:path"); + +// Must match readQueueData in src/lib/local-api.js. If that key ever changes, +// this one has to change with it — a test asserts they agree on real rows. +function queueRowKey(row) { + return `${row.source || ""}|${row.model || ""}|${row.hour_start || ""}`; +} + +const TOKEN_COLUMNS = [ + "input_tokens", + "output_tokens", + "cache_creation_input_tokens", + "cached_input_tokens", + "reasoning_output_tokens", +]; + +const THIRTY_MINUTES_MS = 30 * 60 * 1000; + +// Codex and every-code report reasoning tokens that are ALREADY COUNTED inside +// output_tokens, so their total_tokens correctly excludes the reasoning column — +// adding it would count those tokens twice. +// +// This is not a special case invented here. `computeRowCost` in +// src/lib/pricing/index.js:309 makes exactly the same distinction, and charges +// reasoning at zero for these two sources for the same reason. +// +// Found by running this check against a real 34,922-row queue: 8,236 rows +// "violated" the invariant, every one of them source=codex, and in every case +// the difference was exactly reasoning_output_tokens. The rows were right, the +// check was wrong, and so was the prose it came from — CLAUDE.md said "sum of +// all columns" with no exception. Corrected there too. +const REASONING_FOLDED_INTO_OUTPUT = new Set(["codex", "every-code"]); + +function expectedTotal(row) { + const folded = REASONING_FOLDED_INTO_OUTPUT.has(String(row.source || "").toLowerCase()); + return TOKEN_COLUMNS.reduce((acc, column) => { + if (folded && column === "reasoning_output_tokens") return acc; + const value = Number(row[column] ?? 0); + return acc + (Number.isFinite(value) ? value : 0); + }, 0); +} + +// Decides which lines survive, without touching the disk. Split out so the +// decision is testable on its own and so `analyze` and `compact` cannot drift. +// +// Malformed lines are KEPT. They are invisible to every reader already, so +// dropping them would not change a single API response — but it would destroy +// bytes nobody has looked at, and a partial write worth investigating is +// exactly the kind of thing that should survive a routine maintenance command. +function planCompaction(raw) { + const lines = raw.split("\n"); + const keep = new Set(); + const lastForKey = new Map(); + let parseable = 0; + let malformed = 0; + + lines.forEach((line, index) => { + if (!line.trim()) return; + let row; + try { + row = JSON.parse(line); + } catch { + malformed += 1; + keep.add(index); + return; + } + parseable += 1; + lastForKey.set(queueRowKey(row), index); + }); + + for (const index of lastForKey.values()) keep.add(index); + + const kept = [...keep].sort((a, b) => a - b); + return { + lines: kept.map((index) => lines[index]), + stats: { + totalLines: parseable + malformed, + parseable, + malformed, + uniqueKeys: lastForKey.size, + superseded: parseable - lastForKey.size, + keptLines: kept.length, + }, + }; +} + +function analyzeQueue(queuePath) { + let raw; + try { + raw = fs.readFileSync(queuePath, "utf8"); + } catch (e) { + if (e?.code === "ENOENT") { + return { totalLines: 0, parseable: 0, malformed: 0, uniqueKeys: 0, superseded: 0, keptLines: 0, ratio: 0, bytes: 0 }; + } + throw e; + } + const { stats } = planCompaction(raw); + return { + ...stats, + ratio: stats.parseable > 0 ? stats.superseded / stats.parseable : 0, + bytes: Buffer.byteLength(raw, "utf8"), + }; +} + +// Writes to a temp file in the same directory and renames over the original. +// Same atomic-replace pattern the project-queue rewrite already uses. An +// interrupt between write and rename leaves the original untouched — the temp +// file is the only casualty. +// +// The CALLER holds the sync lock. This does not take it, because the lock is +// per-invocation state owned by the sync command, and a second acquisition +// inside would deadlock against the first. +function compactQueue(queuePath) { + let raw; + try { + raw = fs.readFileSync(queuePath, "utf8"); + } catch (e) { + if (e?.code === "ENOENT") return { changed: false, reason: "no queue file" }; + throw e; + } + + const before = Buffer.byteLength(raw, "utf8"); + const { lines, stats } = planCompaction(raw); + if (stats.superseded === 0) { + return { changed: false, reason: "nothing superseded", ...stats, bytesBefore: before, bytesAfter: before }; + } + + const out = lines.join("\n") + "\n"; + const tmp = path.join( + path.dirname(queuePath), + `${path.basename(queuePath)}.compact.${process.pid}.tmp`, + ); + fs.writeFileSync(tmp, out, "utf8"); + try { + fs.renameSync(tmp, queuePath); + } catch (e) { + fs.unlinkSync(tmp); + throw e; + } + + return { + changed: true, + ...stats, + bytesBefore: before, + bytesAfter: Buffer.byteLength(out, "utf8"), + }; +} + +// CLAUDE.md states the column invariant in prose: +// +// total = input + output + cache_creation + cache_read + reasoning +// +// Nothing enforced it at runtime. A miswritten or corrupt row was aggregated and +// rendered, not flagged — and a parser bug of exactly this shape is the class +// CLAUDE.md records at 1.6-7x magnitude. Same conversion as the curated-expiry +// and version-lockstep checks: a rule that lived in a document starts running. +// +// Returns one finding per violating row, capped by the caller. +function findRowViolations(rows) { + const findings = []; + rows.forEach((row, index) => { + const where = `row ${index + 1} (${row.source || "?"}|${row.model || "?"}|${row.hour_start || "?"})`; + + for (const column of TOKEN_COLUMNS) { + const value = Number(row[column] ?? 0); + if (!Number.isFinite(value)) { + findings.push(`${where}: ${column} is not a number (${JSON.stringify(row[column])})`); + } else if (value < 0) { + findings.push(`${where}: ${column} is negative (${value})`); + } + } + + const sum = expectedTotal(row); + const total = Number(row.total_tokens ?? 0); + if (Number.isFinite(total) && total !== sum) { + findings.push(`${where}: total_tokens ${total} != expected ${sum}`); + } + + const bucket = Date.parse(row.hour_start); + if (!Number.isFinite(bucket)) { + findings.push(`${where}: hour_start is not a timestamp`); + } else if (bucket % THIRTY_MINUTES_MS !== 0) { + findings.push(`${where}: hour_start is not on a 30-minute UTC boundary`); + } + }); + return findings; +} + +module.exports = { + queueRowKey, + expectedTotal, + REASONING_FOLDED_INTO_OUTPUT, + planCompaction, + analyzeQueue, + compactQueue, + findRowViolations, + TOKEN_COLUMNS, + THIRTY_MINUTES_MS, +}; diff --git a/test/init-uninstall.test.js b/test/init-uninstall.test.js index 1b636c64..017bbe65 100644 --- a/test/init-uninstall.test.js +++ b/test/init-uninstall.test.js @@ -33,7 +33,14 @@ function flattenHookEntries(entries) { return entries.flatMap((entry) => (Array.isArray(entry?.hooks) ? entry.hooks : [entry])); } -async function runGeneratedNotifyHandler({ trackerDir, notify }) { +// `home` is REQUIRED, and is the whole point of this signature. The handler is +// run in a CHILD PROCESS, so overriding process.env.HOME in the parent does not +// reach it — and without an isolated HOME the child resolves ~/.tokentracker and +// runs a real sync against the developer's own data. It really did: this test +// appended to and (once queue compaction landed) rewrote an 11 MB production +// queue during `npm test`. +async function runGeneratedNotifyHandler({ trackerDir, notify, home }) { + if (!home) throw new Error("runGeneratedNotifyHandler needs an isolated home"); await fs.mkdir(trackerDir, { recursive: true }); const notifyPath = path.join(trackerDir, "notify.cjs"); await fs.writeFile( @@ -48,7 +55,7 @@ async function runGeneratedNotifyHandler({ trackerDir, notify }) { "utf8", ); await new Promise((resolve, reject) => { - const env = { ...process.env }; + const env = { ...process.env, HOME: home, USERPROFILE: home }; delete env.TOKENTRACKER_DEVICE_TOKEN; const child = require("node:child_process").execFile( process.execPath, @@ -89,12 +96,14 @@ test("notify handler skips SkyComputerUseClient and stale explicit original noti await fs.chmod(skyPath, 0o755); await runGeneratedNotifyHandler({ + home: tmp, trackerDir: path.join(tmp, "tracker-sky"), notify: [skyPath, "turn-ended"], }); assert.equal(await waitForFile(markerPath, { timeoutMs: 500 }), null); await runGeneratedNotifyHandler({ + home: tmp, trackerDir: path.join(tmp, "tracker-missing"), notify: [path.join(tmp, "missing-notify"), "turn-ended"], }); @@ -115,6 +124,7 @@ test("notify handler still chains normal original notify commands", async () => ); await runGeneratedNotifyHandler({ + home: tmp, trackerDir: path.join(tmp, "tracker-safe"), notify: [process.execPath, shimPath], }); @@ -875,3 +885,15 @@ test("init installs Opencode plugin when config dir is missing", async () => { await fs.rm(tmp, { recursive: true, force: true }); } }); + +test("the notify-handler helper refuses to run without an isolated home", async () => { + // A durable guard, not a one-off check. The handler runs in a CHILD process, + // so overriding process.env.HOME in the parent does not reach it. Without this + // the child resolved ~/.tokentracker and ran a real sync against the + // developer's own data — it appended to, and once queue compaction landed + // rewrote, an 11 MB production queue during `npm test`. + await assert.rejects( + () => runGeneratedNotifyHandler({ trackerDir: "/tmp/whatever", notify: ["echo"] }), + /isolated home/, + ); +}); diff --git a/test/parser-conformance.test.js b/test/parser-conformance.test.js index 4da4781a..fb1984aa 100644 --- a/test/parser-conformance.test.js +++ b/test/parser-conformance.test.js @@ -34,6 +34,7 @@ const path = require("node:path"); const { test } = require("node:test"); const rollout = require("../src/lib/rollout"); +const { expectedTotal } = require("../src/lib/queue-compact"); const FIXTURE_DIR = path.join(__dirname, "fixtures", "parser-conformance"); const ALLOWLIST = require("./fixtures/parser-conformance/allowlist.json"); @@ -86,11 +87,15 @@ function assertRowConforms(row, label) { } // CLAUDE.md's "Token normalization" block, which lived in prose and nothing // enforced. A miswritten row was aggregated and rendered, not flagged. - const sum = COLUMNS.reduce((acc, column) => acc + Number(row[column] || 0), 0); + // Not a plain sum: codex and every-code fold reasoning into output_tokens, so + // adding the column again would double-count. expectedTotal() is the single + // rule, shared with doctor's queue.row_invariant check, and it mirrors + // computeRowCost (pricing/index.js:309). Running the plain sum against a real + // 34,922-row queue reported 8,236 false violations, all source=codex. assert.equal( Number(row.total_tokens || 0), - sum, - `${label}: total_tokens must equal the sum of the columns`, + expectedTotal(row), + `${label}: total_tokens must equal the expected total for its source`, ); assert.ok( diff --git a/test/queue-compact.test.js b/test/queue-compact.test.js new file mode 100644 index 00000000..dd71bc81 --- /dev/null +++ b/test/queue-compact.test.js @@ -0,0 +1,229 @@ +"use strict"; + +const assert = require("node:assert/strict"); +const fs = require("node:fs"); +const os = require("node:os"); +const path = require("node:path"); +const { test } = require("node:test"); + +const { + queueRowKey, + planCompaction, + analyzeQueue, + compactQueue, + findRowViolations, + expectedTotal, +} = require("../src/lib/queue-compact"); + +function tmpQueue(rows) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "tt-compact-")); + const queuePath = path.join(dir, "queue.jsonl"); + fs.writeFileSync( + queuePath, + rows.map((r) => (typeof r === "string" ? r : JSON.stringify(r))).join("\n") + "\n", + ); + return queuePath; +} + +const row = (over = {}) => ({ + source: "claude", + model: "claude-sonnet-5", + hour_start: "2026-05-14T09:00:00.000Z", + input_tokens: 100, + output_tokens: 50, + cache_creation_input_tokens: 0, + cached_input_tokens: 0, + reasoning_output_tokens: 0, + total_tokens: 150, + ...over, +}); + +// --- Compaction --------------------------------------------------------------- + +test("compaction keeps the LAST row per key, which is what every reader computes", () => { + const queuePath = tmpQueue([ + row({ input_tokens: 1, output_tokens: 0, total_tokens: 1 }), + row({ input_tokens: 2, output_tokens: 0, total_tokens: 2 }), + row({ input_tokens: 3, output_tokens: 0, total_tokens: 3 }), + ]); + const result = compactQueue(queuePath); + assert.equal(result.changed, true); + assert.equal(result.keptLines, 1); + assert.equal(result.superseded, 2); + const left = fs.readFileSync(queuePath, "utf8").trim().split("\n").map(JSON.parse); + assert.equal(left.length, 1); + assert.equal(left[0].input_tokens, 3, "the surviving row must be the latest, not the first"); +}); + +test("the surviving bytes are the ORIGINAL bytes, not a re-serialised row", () => { + // This is what makes "byte-identical API responses" true by construction. A + // re-serialised row could reorder keys, drop an unknown field a future reader + // needs, or change number formatting. + const original = + '{"source":"claude","model":"m","hour_start":"2026-05-14T09:00:00.000Z","total_tokens":5,"input_tokens":5,"output_tokens":0,"an_unknown_future_field":{"z":1,"a":2}}'; + const queuePath = tmpQueue([JSON.stringify(row()), original]); + compactQueue(queuePath); + const lines = fs.readFileSync(queuePath, "utf8").trim().split("\n"); + assert.ok(lines.includes(original), "the exact original line must survive verbatim"); +}); + +test("what the readers see is unchanged — the real test, not the line count", () => { + // #103's definition of done. Reproduces readQueueData's dedup (same key, + // keep-last) and asserts the result is identical before and after. + const readerView = (queuePath) => { + const seen = new Map(); + for (const line of fs.readFileSync(queuePath, "utf8").split("\n")) { + if (!line.trim()) continue; + let parsed; + try { + parsed = JSON.parse(line); + } catch { + continue; + } + seen.set(queueRowKey(parsed), parsed); + } + return JSON.stringify([...seen.values()]); + }; + + const rows = []; + for (let i = 1; i <= 40; i += 1) { + rows.push(row({ input_tokens: i, total_tokens: i + 50 })); + rows.push( + row({ + source: "codex", + model: "gpt-5.5", + hour_start: "2026-05-14T09:30:00.000Z", + input_tokens: i * 2, + output_tokens: 10, + reasoning_output_tokens: 4, + total_tokens: i * 2 + 10, + }), + ); + } + const queuePath = tmpQueue(rows); + + const before = readerView(queuePath); + const result = compactQueue(queuePath); + const after = readerView(queuePath); + + assert.equal(before, after, "compaction must not change a single byte of what readers compute"); + assert.equal(result.keptLines, 2, "80 rows across 2 keys collapse to 2"); +}); + +test("an unparseable line is kept, not quietly destroyed", () => { + // It is invisible to every reader already, so dropping it would change no API + // response — but a partial write worth investigating should survive a routine + // maintenance command. + const queuePath = tmpQueue([row({ input_tokens: 1, total_tokens: 51 }), "{ truncated par", row()]); + const result = compactQueue(queuePath); + assert.equal(result.malformed, 1); + assert.ok( + fs.readFileSync(queuePath, "utf8").includes("{ truncated par"), + "the unreadable bytes must still be there", + ); +}); + +test("compaction is a no-op when nothing is superseded", () => { + const queuePath = tmpQueue([row(), row({ model: "other-model", total_tokens: 150 })]); + const beforeBytes = fs.readFileSync(queuePath); + const result = compactQueue(queuePath); + assert.equal(result.changed, false); + assert.deepEqual(fs.readFileSync(queuePath), beforeBytes, "the file must not be rewritten"); +}); + +test("an interrupt between write and rename leaves the original intact", () => { + // The reason this uses write-then-rename rather than writing in place. There + // is no way to observe a half-written queue: the rename is atomic, so the file + // is either entirely the old one or entirely the new one. + const queuePath = tmpQueue([row({ input_tokens: 1, total_tokens: 51 }), row()]); + const originalBytes = fs.readFileSync(queuePath); + + const realRename = fs.renameSync; + fs.renameSync = () => { + throw Object.assign(new Error("interrupted"), { code: "EIO" }); + }; + try { + assert.throws(() => compactQueue(queuePath), /interrupted/); + } finally { + fs.renameSync = realRename; + } + + assert.deepEqual(fs.readFileSync(queuePath), originalBytes, "the original must be untouched"); + const leftovers = fs.readdirSync(path.dirname(queuePath)).filter((f) => f.includes(".compact.")); + assert.deepEqual(leftovers, [], "the temp file must be cleaned up, not left behind"); +}); + +test("analyze reports the ratio without touching the file", () => { + const queuePath = tmpQueue([ + row({ input_tokens: 1, total_tokens: 51 }), + row({ input_tokens: 2, total_tokens: 52 }), + row({ model: "other", total_tokens: 150 }), + ]); + const bytes = fs.readFileSync(queuePath); + const stats = analyzeQueue(queuePath); + assert.equal(stats.parseable, 3); + assert.equal(stats.uniqueKeys, 2); + assert.equal(stats.superseded, 1); + assert.ok(Math.abs(stats.ratio - 1 / 3) < 1e-9); + assert.deepEqual(fs.readFileSync(queuePath), bytes); +}); + +test("a missing queue is not an error", () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "tt-compact-")); + const missing = path.join(dir, "queue.jsonl"); + assert.equal(analyzeQueue(missing).parseable, 0); + assert.equal(compactQueue(missing).changed, false); +}); + +test("planCompaction and analyzeQueue cannot drift", () => { + const raw = [row(), row({ input_tokens: 9, total_tokens: 59 })] + .map((r) => JSON.stringify(r)) + .join("\n"); + const queuePath = tmpQueue([row(), row({ input_tokens: 9, total_tokens: 59 })]); + assert.equal(planCompaction(raw).stats.superseded, analyzeQueue(queuePath).superseded); +}); + +// --- The row invariant --------------------------------------------------------- + +test("the invariant flags a column-sum violation, a bad bucket and a negative count", () => { + assert.deepEqual(findRowViolations([row()]), [], "a clean row produces no findings"); + + const cases = { + "total_tokens .* != expected": row({ total_tokens: 151 }), + "is negative": row({ input_tokens: -5, total_tokens: 45 }), + "not on a 30-minute UTC boundary": row({ hour_start: "2026-05-14T09:07:00.000Z" }), + "is not a number": row({ output_tokens: "fifty" }), + }; + for (const [pattern, bad] of Object.entries(cases)) { + const findings = findRowViolations([bad]); + assert.ok( + findings.some((f) => new RegExp(pattern).test(f)), + `expected a finding matching /${pattern}/, got ${JSON.stringify(findings)}`, + ); + } +}); + +test("codex and every-code fold reasoning into output, and are not flagged for it", () => { + // Found by running this check against a real 34,922-row queue: 8,236 rows + // "violated" the invariant, every one source=codex, and in every case the + // difference was exactly reasoning_output_tokens. The rows were right; the + // check and the prose it came from were wrong. computeRowCost + // (pricing/index.js:309) already made this distinction — it charges their + // reasoning at zero for the same reason. + const codexRow = row({ + source: "codex", + model: "gpt-5.5", + input_tokens: 100, + output_tokens: 50, + reasoning_output_tokens: 20, + total_tokens: 150, // reasoning NOT added: it is already inside output_tokens + }); + assert.deepEqual(findRowViolations([codexRow]), []); + assert.equal(expectedTotal(codexRow), 150); + + // The same numbers from a source that does NOT fold must add up to 170, so the + // exception is scoped and not a blanket loosening. + const claudeRow = { ...codexRow, source: "claude" }; + assert.equal(expectedTotal(claudeRow), 170); + assert.ok(findRowViolations([claudeRow]).some((f) => f.includes("!= expected 170"))); +});