diff --git a/__tests__/agent-claude-usage.test.js b/__tests__/agent-claude-usage.test.js index 5ad23fb..3458dc1 100644 --- a/__tests__/agent-claude-usage.test.js +++ b/__tests__/agent-claude-usage.test.js @@ -644,6 +644,75 @@ describe("frozen days", () => { expect(r.projects).toEqual([{ name: "alpha", work: 0, cacheRead: 1, requests: 1 }]); expect(Object.keys(r.openDays)).toHaveLength(29); }); + + // #54: a request whose lines straddle midnight. The day freezes at D+2, so + // the beat that froze it saw every line of the request and gave it to the + // day of its best line. Later beats must make the same choice, not count + // the request again on the open day once the frozen line is filtered out. + // Toronto is UTC-4 here: 03:59Z is Oct 1 23:59, 04:01Z is Oct 2 00:01. + it("a request straddling midnight whose best line sits on a frozen day is not counted again on the open day", async () => { + writeTranscript( + "a/straddle.jsonl", + [ + line({ ts: "2026-10-02T03:59:00Z", id: "msg_mid", req: "req_mid", output: 5 }), + line({ ts: "2026-10-02T04:01:00Z", id: "msg_mid", req: "req_mid", output: 5 }), + ], + "2026-10-02T12:00:00Z", + ); + const r = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME, frozenDays: FROZEN }); + expect(dayOf(r, "2026-10-02").models).toEqual({}); + expect(r.openDays["2026-10-02"].models).toEqual({}); + // The frozen day is served as it was counted, untouched. + expect(dayOf(r, "2026-10-01").models["claude-opus-5-5"].requests).toBe(7); + }); + + it("a request straddling midnight whose best line sits on the open day is counted once there", async () => { + writeTranscript( + "a/straddle.jsonl", + [ + line({ ts: "2026-10-02T03:59:00Z", id: "msg_mid", req: "req_mid", output: 1 }), + line({ ts: "2026-10-02T04:01:00Z", id: "msg_mid", req: "req_mid", output: 5 }), + ], + "2026-10-02T12:00:00Z", + ); + const r = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME, frozenDays: FROZEN }); + expect(dayOf(r, "2026-10-02").models["claude-opus-5-5"]).toEqual({ + input: 1, + output: 5, + cacheCreation: 0, + cacheRead: 0, + requests: 1, + }); + }); + + it("on equal output the earliest line wins, whatever the order the files are read in", async () => { + // Sorted read order puts the later line first. The beat that freezes a day + // and the beats after it do not read the same files (the threshold moves), + // so the winner must not depend on the read order. + const later = writeTranscript( + "a/a-first.jsonl", + [line({ ts: "2026-10-02T04:01:00Z", id: "msg_tie", req: "req_tie", output: 5 })], + "2026-10-02T12:00:00Z", + ); + const earlier = writeTranscript( + "a/b-second.jsonl", + [line({ ts: "2026-10-02T03:59:00Z", id: "msg_tie", req: "req_tie", output: 5 })], + "2026-10-02T12:00:00Z", + ); + const opened = []; + const r = await aggregate({ + root: tmpDir, + today: "2026-10-04", + homedir: HOME, + openFile: (f, o) => { + opened.push(f); + return fs.createReadStream(f, o); + }, + }); + expect(opened).toEqual([later, earlier]); + expect(dayOf(r, "2026-10-01").models["claude-opus-5-5"].requests).toBe(1); + expect(dayOf(r, "2026-10-02").models).toEqual({}); + }); }); // --- Robustness ----------------------------------------------------------------- @@ -684,6 +753,70 @@ describe("robustness", () => { }); }); + // #54: an unreadable transcript, directory or stat is counted, so that the + // agent freezes nothing on that beat. + it("counts a stream error as a read error", async () => { + writeTranscript("p/good.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z"); + const bad = writeTranscript("p/bad.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z"); + const r = await aggregate({ + root: tmpDir, + today: "2026-10-04", + homedir: HOME, + openFile: (f, o) => fs.createReadStream(f === bad ? path.join(tmpDir, "does-not-exist") : f, o), + }); + expect(r.readErrors).toBe(1); + }); + + it("counts an opener that throws as a read error", async () => { + writeTranscript("p/bad.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z"); + const r = await aggregate({ + root: tmpDir, + today: "2026-10-04", + homedir: HOME, + openFile: () => { + throw Object.assign(new Error("denied"), { code: "EACCES" }); + }, + }); + expect(r.readErrors).toBe(1); + }); + + it("counts an unreadable subdirectory as a read error, and skips its files", async () => { + writeTranscript("p/good.jsonl", [line({ ts: "2026-10-03T15:00:00Z", output: 4 })], "2026-10-04T12:00:00Z"); + writeTranscript("q/hidden.jsonl", [line({ ts: "2026-10-03T15:00:00Z", output: 9 })], "2026-10-04T12:00:00Z"); + const badDir = path.join(tmpDir, "q"); + // A spy rather than chmod 000: the CI runs as root, which reads anyway. + const realReaddir = fs.promises.readdir; + vi.spyOn(fs.promises, "readdir").mockImplementation((dir, opts) => + dir === badDir + ? Promise.reject(Object.assign(new Error("denied"), { code: "EACCES" })) + : realReaddir.call(fs.promises, dir, opts), + ); + const r = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME }); + expect(r.readErrors).toBe(1); + expect(dayOf(r, "2026-10-03").models["claude-opus-5-5"].output).toBe(4); + }); + + it("counts a failed stat as a read error, ENOENT included", async () => { + writeTranscript("p/good.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z"); + const gone = writeTranscript("p/gone.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z"); + const realStat = fs.promises.stat; + vi.spyOn(fs.promises, "stat").mockImplementation((file, opts) => + file === gone + ? Promise.reject(Object.assign(new Error("gone"), { code: "ENOENT" })) + : realStat.call(fs.promises, file, opts), + ); + const r = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME }); + expect(r.readErrors).toBe(1); + }); + + it("a missing root is an empty scan, not a read error; a clean scan has none", async () => { + const missing = await aggregate({ root: path.join(tmpDir, "absent"), today: "2026-10-04", homedir: HOME }); + expect(missing.readErrors).toBe(0); + writeTranscript("p/good.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z"); + const clean = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME }); + expect(clean.readErrors).toBe(0); + }); + it("folds a model name outside the server key pattern into `other`", async () => { writeTranscript( "p/s.jsonl", @@ -995,6 +1128,61 @@ describe("limits — the endpoint's answer", () => { }); }); +describe("limits — the response size is counted in bytes, while reading (#54)", () => { + const MAX = 256 * 1024; + const limitsOpts = (fetch) => ({ + env: {}, + credentialsPath: "/unused", + readFile: () => JSON.stringify(goodCredentials()), + claudeCodeVersion: "2.1.283", + now: NOW, + fetch, + }); + // A valid answer padded to exactly `bytes` UTF-8 bytes with `pad`. + function padded(bytes, pad) { + const base = JSON.stringify({ ...usageFixture(), pad: "" }); + let room = bytes - Buffer.byteLength(base); + const unit = Buffer.byteLength(pad); + let fill = "a".repeat(room % unit); + fill += pad.repeat(Math.floor(room / unit)); + const body = JSON.stringify({ ...usageFixture(), pad: fill }); + expect(Buffer.byteLength(body)).toBe(bytes); + return body; + } + + it("accepts a body of exactly 256 KiB", async () => { + const body = padded(MAX, "a"); + const r = await fetchLimits(limitsOpts(makeFetch(() => new Response(body, { status: 200 })))); + expect(r.status).toBe("ok"); + }); + + it("refuses 256 KiB + 1 byte of two-byte characters, fewer than 256 Ki characters", async () => { + const body = padded(MAX + 1, "\u00e9"); + expect(body.length).toBeLessThan(MAX); + const r = await fetchLimits(limitsOpts(makeFetch(() => new Response(body, { status: 200 })))); + expect(r).toEqual({ status: "unavailable", detail: DETAILS.BAD_RESPONSE, fetchedAt: null, items: [] }); + }); + + it("stops reading an endless body and cancels it", async () => { + let pulls = 0; + let cancelled = false; + const chunk = new Uint8Array(64 * 1024).fill(0x20); + const stream = new ReadableStream({ + pull(controller) { + pulls += 1; + controller.enqueue(chunk); + }, + cancel() { + cancelled = true; + }, + }); + const r = await fetchLimits(limitsOpts(makeFetch(() => new Response(stream, { status: 200 })))); + expect(r).toEqual({ status: "unavailable", detail: DETAILS.BAD_RESPONSE, fetchedAt: null, items: [] }); + expect(cancelled).toBe(true); + expect(pulls).toBeLessThan(10); // 5 chunks cross the ceiling + }); +}); + // --- limits: when no call is made ---------------------------------------------- describe("limits — cases that never call the endpoint", () => { @@ -1252,6 +1440,24 @@ describe("state — local cache", () => { expect(Object.keys(s.frozenDays).sort()).toEqual(["2026-08-30", "2026-09-20"]); }); + it("with freeze: false, keeps the frozen days (pruned) and freezes no new one", () => { + const state = { + ...emptyState(), + frozenDays: { + "2026-08-29": { models: {}, projects: {} }, // pruned all the same + "2026-09-20": { models: {}, projects: {} }, + }, + }; + const openDays = { + "2026-10-01": { models: { m: { input: 1 } }, projects: {} }, + "2026-10-03": { models: {}, projects: {} }, + }; + const ok = { status: "ok", fetchedAt: "2026-10-04T16:00:00.000Z", items: [{ kind: "session" }] }; + const s = nextState(state, { openDays, today: "2026-10-04", limits: ok, freeze: false }); + expect(Object.keys(s.frozenDays)).toEqual(["2026-09-20"]); + expect(s.lastGoodLimits).toEqual({ fetchedAt: ok.fetchedAt, items: ok.items }); + }); + it("updates lastGoodLimits on `ok` only", () => { const ok = { status: "ok", fetchedAt: "2026-10-04T16:00:00.000Z", items: [{ kind: "session" }] }; const s1 = nextState(emptyState(), { openDays: {}, today: "2026-10-04", limits: ok }); @@ -1292,6 +1498,19 @@ describe("state — local cache", () => { expect(loadState(file).frozenDays).toEqual({ "2026-10-01": { models: {}, projects: {} } }); }); + it("a temp file left at mode 644 by a killed run still ends in a mode 600 cache", () => { + // writeFileSync only applies `mode` when it creates the file (#54). + const file = path.join(tmpDir, "state", "state.json"); + fs.mkdirSync(path.dirname(file), { recursive: true }); + const tmp = `${file}.${process.pid}.tmp`; + fs.writeFileSync(tmp, "leftover"); + fs.chmodSync(tmp, 0o644); // explicit: the umask could already give 600 + expect(fs.statSync(tmp).mode & 0o777).toBe(0o644); + saveState(file, emptyState()); + expect(fs.existsSync(tmp)).toBe(false); + expect(fs.statSync(file).mode & 0o777).toBe(0o600); + }); + it("the default state dir is ~/.local/state/maximus-claude-usage", () => { expect(stateFilePath(defaultStateDir("/home/u"))).toBe("/home/u/.local/state/maximus-claude-usage/state.json"); }); @@ -1353,6 +1572,42 @@ describe("push-claude-usage — one beat", () => { expect(snap2.projects).toEqual(snap1.projects); }); + it("a beat with a read error freezes no day, still pushes, logs a count only; the next clean beat freezes", async () => { + const { home, stateFile } = makeHome({ credentials: undefined }); + const projectDir = path.join(home, ".claude", "projects", "-home-tester-claude-code-alpha"); + fs.mkdirSync(projectDir, { recursive: true }); + const write = (name, lines) => { + const file = path.join(projectDir, name); + fs.writeFileSync(file, `${lines.map((l) => JSON.stringify(l)).join("\n")}\n`); + setMtime(file, "2026-10-04T15:30:00Z"); + return file; + }; + write("good.jsonl", [line({ ts: "2026-10-01T15:00:00Z", output: 3 })]); + const bad = write("secret-name.jsonl", [line({ ts: "2026-10-01T15:00:00Z", output: 8 })]); + const env = agentEnv({ CLAUDE_USAGE_LIMITS: "off" }); + + const fetch1 = routedFetch(); + const r1 = await runAgent({ + home, fetch: fetch1, env, + openFile: (f, o) => fs.createReadStream(f === bad ? path.join(projectDir, "missing.jsonl") : f, o), + }); + expect(r1.code).toBe(EXIT_OK); + expect(pushCalls(fetch1)).toHaveLength(1); + expect(loadState(stateFile).frozenDays).toEqual({}); + expect(r1.stderr).toContain("read_errors=1"); + expect(r1.stderr).not.toContain("secret-name"); + expect(r1.stderr).not.toContain(home); + expect(r1.stdout).not.toContain("secret-name"); + + const fetch2 = routedFetch(); + const r2 = await runAgent({ home, fetch: fetch2, env, now: NOW + 15 * 60 * 1000 }); + expect(r2.code).toBe(EXIT_OK); + expect(r2.stderr).not.toContain("read_errors"); + const frozen = loadState(stateFile).frozenDays; + expect(Object.keys(frozen)).toHaveLength(28); + expect(frozen["2026-10-01"].models["claude-opus-5-5"].output).toBe(11); + }); + it("a corrupt state.json gives a full beat without failing, and is rewritten", async () => { const { home, stateFile } = makeHome({ fixtures: true }); fs.mkdirSync(path.dirname(stateFile), { recursive: true }); diff --git a/agent/claude-usage/aggregate.js b/agent/claude-usage/aggregate.js index d0bf34e..24d6c62 100644 --- a/agent/claude-usage/aggregate.js +++ b/agent/claude-usage/aggregate.js @@ -16,7 +16,19 @@ // times (one line per content block, and again in subagent transcripts), // with identical input/cache counters and a growing output_tokens. Lines // are deduplicated on (message.id, requestId) across ALL files, keeping the -// line with the largest output_tokens. +// line with the largest output_tokens; on equal output, the earliest line, +// so the choice never depends on the order the files are read in. +// * FROZEN DAYS ARE DROPPED AFTER DEDUPLICATION, NOT BEFORE. A request whose +// lines straddle midnight belongs to the day of its best line. The beat +// that froze a day saw all its lines (a day freezes at D+2) and made that +// choice; filtering the frozen day's lines first would leave the later +// line alone and count the request a second time on the open day (#54). +// Known limit: the window start is still filtered per line, so a request +// straddling the window's first midnight lands on the window's first day. +// * READ ERRORS ARE COUNTED. An unreadable directory, a failed stat or a +// stream error skips that input but is reported in `readErrors`: the +// agent then freezes nothing on that beat, rather than freezing a day +// with lines missing (#54). A missing root is an empty scan, not an error. // * THE SCAN THRESHOLD. Frozen days are never recomputed. A file whose mtime // (Toronto day) is before the oldest open day of the window cannot hold a // line on an open day, so it is not even opened: this is what keeps a run @@ -225,8 +237,9 @@ function normalizeDay(day) { // --- Transcript scan -------------------------------------------------------- // Every *.jsonl under root, recursively (subagents/ included). Symlinks are -// neither followed nor read. A missing root is an empty scan, not an error. -async function listTranscripts(root) { +// neither followed nor read. A missing root is an empty scan, not an error; +// any other directory that cannot be listed is reported through onError. +async function listTranscripts(root, onError = () => {}) { const out = []; const stack = [root]; while (stack.length) { @@ -234,7 +247,8 @@ async function listTranscripts(root) { let entries; try { entries = await fs.promises.readdir(dir, { withFileTypes: true }); - } catch { + } catch (err) { + if (!(dir === root && err && err.code === "ENOENT")) onError(); continue; } for (const e of entries) { @@ -247,33 +261,41 @@ async function listTranscripts(root) { return out; } -// Calls onLine for each line of the file. Resolves once the file is consumed; -// a read error (file removed mid-run, permission) resolves too: one unreadable -// transcript must not sink the whole snapshot. +// Calls onLine for each line of the file. Resolves once the file is consumed, +// with true when it was read to the end; a read error (file removed mid-run, +// permission) resolves with false: one unreadable transcript must not sink +// the whole snapshot, but the caller must know the scan is incomplete. function readLines(file, openFile, onLine) { return new Promise((resolve) => { let stream; try { stream = openFile(file, { encoding: "utf-8" }); } catch { - resolve(); + resolve(false); return; } let done = false; let rl = null; + // The flag is set by the error handlers only, before finish(): neither a + // 'close' after an error nor the destroy() below can turn it back. + let failed = false; const finish = () => { if (done) return; done = true; if (rl) rl.close(); if (typeof stream.destroy === "function") stream.destroy(); - resolve(); + resolve(!failed); + }; + const fail = () => { + failed = true; + finish(); }; // Listen on both: readline forwards the input's error to the interface, // and an unlistened 'error' there is an uncaught exception that kills the // process (and the interface then never emits 'close'). - stream.on("error", finish); + stream.on("error", fail); rl = readline.createInterface({ input: stream, crlfDelay: Infinity }); - rl.on("error", finish); + rl.on("error", fail); rl.on("line", onLine); rl.on("close", finish); }); @@ -294,6 +316,7 @@ function readLines(file, openFile, onLine) { * days: Array<{ date: string, models: Object }>, * projects: Array<{ name: string, work: number, cacheRead: number, requests: number }>, * claudeCodeVersion: string | null, + * readErrors: number, unreadable directories, stats and transcripts; 0 on a complete scan * }>} */ async function aggregate({ @@ -357,9 +380,11 @@ async function aggregate({ const day = torontoDay(ts); if (day < windowStart || day > today) return; - if (Object.prototype.hasOwnProperty.call(frozen, day)) return; + // Frozen days are NOT filtered here: see FROZEN DAYS ARE DROPPED AFTER + // DEDUPLICATION in the header. const entry = { + ts, day, model: typeof model === "string" && MODEL_KEY_RE.test(model) ? model : OTHER_MODEL, project: mapProject(obj.cwd, homedir), @@ -375,18 +400,27 @@ async function aggregate({ // it counts as its own request rather than being dropped. const key = id && reqId ? `${id}\u0000${reqId}` : `\u0001${anonymous++}`; const prev = requests.get(key); - if (!prev || entry.output > prev.output) requests.set(key, entry); + if (!prev || entry.output > prev.output || (entry.output === prev.output && entry.ts < prev.ts)) { + requests.set(key, entry); + } }; - for (const file of await listTranscripts(root)) { + let readErrors = 0; + const onError = () => { + readErrors += 1; + }; + for (const file of await listTranscripts(root, onError)) { let stat; try { stat = await fs.promises.stat(file); } catch { + // ENOENT included: a transcript removed between the walk and the stat + // may have held lines of a day about to be frozen. + onError(); continue; } if (torontoDay(stat.mtimeMs) < threshold) continue; - await readLines(file, openFile, onLine); + if (!(await readLines(file, openFile, onLine))) onError(); } // Open days: every non-frozen day of the window, empty ones included, so @@ -395,6 +429,8 @@ async function aggregate({ const openDays = dict(); for (const day of openDayList) openDays[day] = { models: dict(), projects: dict() }; for (const r of requests.values()) { + // A request whose best line sits on a frozen day was counted there. + if (Object.prototype.hasOwnProperty.call(frozen, r.day)) continue; const bucket = openDays[r.day]; if (!bucket.models[r.model]) bucket.models[r.model] = emptyModel(); addModel(bucket.models[r.model], { ...r, requests: 1 }); @@ -425,6 +461,7 @@ async function aggregate({ days, projects: topProjects(projectTotals), claudeCodeVersion: latest ? latest.version : null, + readErrors, }; } diff --git a/agent/claude-usage/limits.js b/agent/claude-usage/limits.js index 1ac23fc..a251970 100644 --- a/agent/claude-usage/limits.js +++ b/agent/claude-usage/limits.js @@ -29,7 +29,8 @@ const path = require("node:path"); const USAGE_URL = "https://api.anthropic.com/api/oauth/usage"; const ANTHROPIC_BETA = "oauth-2025-04-20"; const LIMITS_TIMEOUT_MS = 10000; -// Far beyond the ~2 KB the endpoint answers; past it the body is not parsed. +// Far beyond the ~2 KB the endpoint answers. Counted in bytes while reading: +// past it the stream is cancelled, neither read further nor parsed (#54). const MAX_RESPONSE_BYTES = 256 * 1024; const MAX_ITEMS = 12; const MAX_STRING = 128; @@ -175,9 +176,22 @@ function readCredentials(credentialsPath, readFile) { } async function readBody(res) { - const text = await res.text(); - if (text.length > MAX_RESPONSE_BYTES) throw new Error("too large"); - return JSON.parse(text); + if (!res.body || typeof res.body.getReader !== "function") throw new Error("no body"); + const reader = res.body.getReader(); + const chunks = []; + let total = 0; + for (;;) { + // A timeout surfaces here as a TimeoutError/AbortError, mapped by the caller. + const { done, value } = await reader.read(); + if (done) break; + total += value.byteLength; + if (total > MAX_RESPONSE_BYTES) { + reader.cancel().catch(() => {}); + throw new Error("too large"); + } + chunks.push(value); + } + return JSON.parse(Buffer.concat(chunks).toString("utf-8")); } function discardBody(res) { diff --git a/agent/claude-usage/push-claude-usage.js b/agent/claude-usage/push-claude-usage.js index 87756c0..412ffa6 100755 --- a/agent/claude-usage/push-claude-usage.js +++ b/agent/claude-usage/push-claude-usage.js @@ -177,7 +177,13 @@ async function main(argv = process.argv.slice(2), env = process.env, deps = {}) now, }); - const updated = nextState(state, { openDays: aggregated.openDays, today, limits }); + // An incomplete scan freezes nothing. Only a count is printed: a path + // would put a project name in syslog for good. + const complete = aggregated.readErrors === 0; + if (!complete) { + logError(`${PREFIX} warning: read_errors=${aggregated.readErrors}, no day frozen this beat`); + } + const updated = nextState(state, { openDays: aggregated.openDays, today, limits, freeze: complete }); // Cache first. A dry run is read-only: it shows what would be sent and // leaves the cache exactly as the cron left it. if (!dryRun) { diff --git a/agent/claude-usage/state.js b/agent/claude-usage/state.js index 7a83ddb..275eaa0 100644 --- a/agent/claude-usage/state.js +++ b/agent/claude-usage/state.js @@ -92,8 +92,9 @@ function loadState(file, { readFile = (f) => fs.readFileSync(f, "utf-8") } = {}) * @param {object} opts.openDays aggregate().openDays * @param {string} opts.today Toronto "YYYY-MM-DD" * @param {object} opts.limits fetchLimits() result + * @param {boolean} [opts.freeze] false: freeze no new day (incomplete scan) */ -function nextState(state, { openDays = {}, today, limits = null }) { +function nextState(state, { openDays = {}, today, limits = null, freeze = true }) { const yesterday = shiftDay(today, -1); const oldest = shiftDay(today, -RETENTION_DAYS); @@ -101,7 +102,10 @@ function nextState(state, { openDays = {}, today, limits = null }) { for (const [day, value] of Object.entries(state.frozenDays || {})) { if (day >= oldest) frozenDays[day] = value; } - for (const [day, value] of Object.entries(openDays || {})) { + // freeze: false after a scan that hit a read error (#54): a day frozen + // with lines missing would stay wrong for RETENTION_DAYS. The next clean + // beat freezes it instead. + for (const [day, value] of Object.entries(freeze ? openDays || {} : {})) { if (!isDay(day) || day >= yesterday || day < oldest) continue; frozenDays[day] = { models: value.models || {}, projects: value.projects || {} }; }