vps-health-api/agent/claude-usage/aggregate.js

// Pure aggregation of Claude Code usage from the local transcripts (issue #49).
//
// Reads ~/.claude/projects/**/*.jsonl (subagent transcripts included) and
// builds the `days` series and the `projects` top of the POST /claude-usage
// contract (schemaVersion 1). No network, no runtime dependency, nothing
// written: the caller (the claude-usage agent) owns the frozen-days state and
// the delivery.
//
// Properties that are load-bearing and easy to break:
//
//   * DATES ARE STRINGS. A day is the America/Toronto calendar date as
//     "YYYY-MM-DD"; shifting by n days goes through Date.UTC on the parts.
//     Never "now - 86 400 000": DST days last 23 or 25 hours, and a
//     millisecond offset lands on the wrong date around the change.
//   * ONE REQUEST, ONE COUNT. Claude Code writes the same API response several
//     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; 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 limits: the window start is still filtered per line, so a request
//     straddling the window's first midnight lands on the window's first day;
//     and the mtime threshold may skip a subagent transcript holding the only
//     copy of a straddling request's earlier line, in which case a later beat
//     sees fewer lines of it than the beat that froze the 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
//     from re-reading ~210 MB of transcripts every time. The threshold comes
//     from the open days, never from the frozen keys: a stray frozen day (in
//     the future, or after a gap) must not hide an open day's files.
//   * NO PROTOTYPE. Names read from transcripts or the state file key plain
//     maps built by dict(), so `__proto__` or `constructor` stay ordinary keys.

const fs = require("node:fs");
const os = require("node:os");
const path = require("node:path");
const readline = require("node:readline");

const TIME_ZONE = "America/Toronto";
const WINDOW_DAYS = 30;
const TOP_PROJECTS = 8;
const MAX_MODELS_PER_DAY = 12;

const OTHER_MODEL = "other";
const OTHER_PROJECTS = "(autres)";
const MONOREPO_PROJECT = "(monorepo)";
const UNKNOWN_PROJECT = "(autre)";
const DEFENSEUR_AUTO_PROJECT = "defenseurs";
const SYNTHETIC_MODEL = "<synthetic>";

// Same bounds as the server's sanitizer: a model key outside this pattern
// would be dropped there, so it is folded into `other` here instead.
const MODEL_KEY_RE = /^[a-z0-9][a-z0-9.\-]{0,63}$/;
const DAY_RE = /^\d{4}-\d{2}-\d{2}$/;
const MAX_STRING = 128;

// --- Dates ------------------------------------------------------------------

const dayFormatter = new Intl.DateTimeFormat("en-CA", {
  timeZone: TIME_ZONE,
  year: "numeric",
  month: "2-digit",
  day: "2-digit",
});

// Toronto calendar day of an instant (Date or epoch ms), as "YYYY-MM-DD".
// Built from formatToParts so the result does not depend on the en-CA
// separator chosen by the ICU build.
function torontoDay(instant) {
  const parts = {};
  for (const p of dayFormatter.formatToParts(instant)) parts[p.type] = p.value;
  return `${parts.year}-${parts.month}-${parts.day}`;
}

// "YYYY-MM-DD" shifted by n calendar days (n may be negative). Pure calendar
// arithmetic on the parts: the time zone and DST never enter into it.
function shiftDay(day, n) {
  const [y, m, d] = day.split("-").map(Number);
  return new Date(Date.UTC(y, m - 1, d + n)).toISOString().slice(0, 10);
}

// The WINDOW_DAYS dates ending on `today`, oldest first.
function windowDays(today) {
  const days = [];
  for (let i = WINDOW_DAYS - 1; i >= 0; i--) days.push(shiftDay(today, -i));
  return days;
}

function isDay(value) {
  if (typeof value !== "string" || !DAY_RE.test(value)) return false;
  // Rejects 2026-02-30 and friends: a real date round-trips unchanged.
  return shiftDay(value, 0) === value;
}

// --- Project mapping --------------------------------------------------------

function escapeRe(s) {
  return s.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}

// Collapses repeated slashes and drops trailing ones, so "<home>/claude-code//x"
// cannot yield an empty project name.
function stripTrailingSlashes(p) {
  const collapsed = p.replace(/\/{2,}/g, "/");
  return collapsed.length > 1 ? collapsed.replace(/\/+$/, "") : collapsed;
}

// Project name from a transcript line's `cwd`, rules applied in order:
//   1. <home>/claude-code/<a>/...      -> <a>  (.spikes and .archive -> <a>/<b>)
//   2. <home>/claude-code exactly      -> (monorepo)
//   3. /tmp/spike-defenseur-auto-*     -> defenseurs
//   4. /tmp/claude-<uid>/<home-slug>-claude-code-<p>/... -> <p>
//   5. anything else                   -> (autre)
function mapProject(cwd, homedir) {
  if (typeof cwd !== "string" || cwd === "") return UNKNOWN_PROJECT;
  const dir = stripTrailingSlashes(cwd);
  const home = stripTrailingSlashes(String(homedir || ""));
  const monorepo = `${home}/claude-code`;

  if (home && dir.startsWith(`${monorepo}/`)) {
    const parts = dir.slice(monorepo.length + 1).split("/");
    const a = parts[0];
    if ((a === ".spikes" || a === ".archive") && parts[1]) {
      return `${a}/${parts[1]}`.slice(0, MAX_STRING);
    }
    return a.slice(0, MAX_STRING);
  }
  if (home && dir === monorepo) return MONOREPO_PROJECT;
  if (dir.startsWith("/tmp/spike-defenseur-auto-")) return DEFENSEUR_AUTO_PROJECT;

  if (home) {
    // Claude Code names its per-project temp dir after the cwd, slashes turned
    // into dashes: /home/max/claude-code/foo -> -home-max-claude-code-foo.
    const slug = `${home.replace(/\//g, "-")}-claude-code-`;
    const re = new RegExp(`^/tmp/claude-\\d+/${escapeRe(slug)}([^/]+)(?:/|$)`);
    const m = re.exec(dir);
    if (m) return m[1].slice(0, MAX_STRING);
  }
  return UNKNOWN_PROJECT;
}

// --- Counters ---------------------------------------------------------------

function count(value) {
  return Number.isSafeInteger(value) && value >= 0 ? value : 0;
}

// Every map keyed by a name read from a transcript or the state file (model,
// project, day) has no prototype: a project named `__proto__` or
// `constructor` is then a plain key, never a write into Object.prototype.
function dict() {
  return Object.create(null);
}

function emptyModel() {
  return { input: 0, output: 0, cacheCreation: 0, cacheRead: 0, requests: 0 };
}

function emptyProject() {
  return { work: 0, cacheRead: 0, requests: 0 };
}

function addModel(target, src) {
  target.input += count(src.input);
  target.output += count(src.output);
  target.cacheCreation += count(src.cacheCreation);
  target.cacheRead += count(src.cacheRead);
  target.requests += count(src.requests);
}

function addProject(target, src) {
  target.work += count(src.work);
  target.cacheRead += count(src.cacheRead);
  target.requests += count(src.requests);
}

function modelWork(m) {
  return m.input + m.output + m.cacheCreation;
}

// At most MAX_MODELS_PER_DAY keys: beyond that, the least consuming models
// (by work tokens, then cache reads, then name for a stable order) are merged
// into `other`. Idempotent: a day already capped comes back unchanged.
function capModels(models) {
  const names = Object.keys(models).filter((k) => k !== OTHER_MODEL);
  const hasOther = Object.prototype.hasOwnProperty.call(models, OTHER_MODEL);
  if (names.length + (hasOther ? 1 : 0) <= MAX_MODELS_PER_DAY) return models;

  names.sort((a, b) => {
    const ma = models[a];
    const mb = models[b];
    return (
      modelWork(mb) - modelWork(ma) ||
      mb.cacheRead - ma.cacheRead ||
      (a < b ? -1 : a > b ? 1 : 0)
    );
  });
  const out = dict();
  for (const name of names.slice(0, MAX_MODELS_PER_DAY - 1)) out[name] = models[name];
  const other = emptyModel();
  if (hasOther) addModel(other, models[OTHER_MODEL]);
  for (const name of names.slice(MAX_MODELS_PER_DAY - 1)) addModel(other, models[name]);
  out[OTHER_MODEL] = other;
  return out;
}

// Frozen days come from the agent's state file: re-read through the same
// counters, never trusted verbatim.
function normalizeDay(day) {
  const models = dict();
  const projects = dict();
  const srcModels = day && typeof day.models === "object" && day.models ? day.models : {};
  const srcProjects =
    day && typeof day.projects === "object" && day.projects ? day.projects : {};
  for (const [name, m] of Object.entries(srcModels)) {
    if (!m || typeof m !== "object") continue;
    const key = MODEL_KEY_RE.test(name) ? name : OTHER_MODEL;
    if (!models[key]) models[key] = emptyModel();
    addModel(models[key], m);
  }
  for (const [name, p] of Object.entries(srcProjects)) {
    if (!p || typeof p !== "object") continue;
    const key = name.slice(0, MAX_STRING);
    if (!projects[key]) projects[key] = emptyProject();
    addProject(projects[key], p);
  }
  return { models, projects };
}

// --- 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;
// any other directory that cannot be listed is reported through onError.
async function listTranscripts(root, onError = () => {}) {
  const out = [];
  const stack = [root];
  while (stack.length) {
    const dir = stack.pop();
    let entries;
    try {
      entries = await fs.promises.readdir(dir, { withFileTypes: true });
    } catch (err) {
      if (!(dir === root && err && err.code === "ENOENT")) onError();
      continue;
    }
    for (const e of entries) {
      const full = path.join(dir, e.name);
      if (e.isDirectory()) stack.push(full);
      else if (e.isFile() && e.name.endsWith(".jsonl")) out.push(full);
    }
  }
  out.sort();
  return out;
}

// 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(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(!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", fail);
    rl = readline.createInterface({ input: stream, crlfDelay: Infinity });
    rl.on("error", fail);
    rl.on("line", onLine);
    rl.on("close", finish);
  });
}

/**
 * Aggregate the transcripts under `root`.
 *
 * @param {object}  opts
 * @param {string}  opts.root        transcripts root (~/.claude/projects)
 * @param {object} [opts.frozenDays] { "YYYY-MM-DD": { models, projects } } — never recomputed
 * @param {string} [opts.today]      Toronto date "YYYY-MM-DD" (default: now)
 * @param {string} [opts.homedir]    home used by the project mapping (default: os.homedir())
 * @param {Function} [opts.openFile] (path, options) => Readable — injectable for tests
 * @returns {Promise<{
 *   openDays: Object<string, { models: Object<string, {input, output, cacheCreation, cacheRead, requests}>,
 *                              projects: Object<string, {work, cacheRead, requests}> }>,
 *   days: Array<{ date: string, models: Object<string, {input, output, cacheCreation, cacheRead, requests}> }>,
 *   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({
  root,
  frozenDays = {},
  today = torontoDay(Date.now()),
  homedir = os.homedir(),
  openFile = (file, options) => fs.createReadStream(file, options),
} = {}) {
  if (!isDay(today)) throw new Error("aggregate: today must be a YYYY-MM-DD date");

  const frozen = dict();
  if (frozenDays && typeof frozenDays === "object") {
    for (const [day, value] of Object.entries(frozenDays)) {
      if (isDay(day)) frozen[day] = value;
    }
  }

  const window = windowDays(today);
  const windowStart = window[0];
  const openDayList = window.filter((d) => !Object.prototype.hasOwnProperty.call(frozen, d));

  // Scan threshold: the oldest open day of the window. With the agent's
  // contiguous frozen prefix this is the day after the most recent frozen day;
  // computed from the open days rather than from the frozen keys, so a frozen
  // day in the future or a gap in the frozen run can never push the threshold
  // past an open day and silently empty it. A file last written before the
  // window cannot hold a line inside it. Every day frozen: nothing to read.
  const threshold = openDayList.length ? openDayList[0] : shiftDay(today, 1);

  // (message.id, requestId) -> best line so far.
  const requests = new Map();
  let anonymous = 0;
  let latest = null; // { ts, version }

  const onLine = (line) => {
    // Cheap prefilter: a useful line always carries "usage". Saves a
    // JSON.parse on the user/tool lines, the bulk of the ~210 MB.
    if (!line.includes('"usage"')) return;
    let obj;
    try {
      obj = JSON.parse(line);
    } catch {
      return; // truncated or garbled line: skipped
    }
    if (!obj || obj.type !== "assistant") return;
    const message = obj.message;
    if (!message || typeof message !== "object") return;
    const usage = message.usage;
    if (!usage || typeof usage !== "object") return;

    const ts = typeof obj.timestamp === "string" ? Date.parse(obj.timestamp) : NaN;
    if (!Number.isFinite(ts)) return;

    if (typeof obj.version === "string" && obj.version && (!latest || ts > latest.ts)) {
      latest = { ts, version: obj.version.slice(0, MAX_STRING) };
    }

    const model = message.model;
    if (model === SYNTHETIC_MODEL) return;

    const day = torontoDay(ts);
    if (day < windowStart || day > today) 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),
      input: count(usage.input_tokens),
      output: count(usage.output_tokens),
      cacheCreation: count(usage.cache_creation_input_tokens),
      cacheRead: count(usage.cache_read_input_tokens),
    };

    const id = typeof message.id === "string" ? message.id : "";
    const reqId = typeof obj.requestId === "string" ? obj.requestId : "";
    // A line without its identifiers cannot be matched with its duplicates:
    // 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 || (entry.output === prev.output && entry.ts < prev.ts)) {
      requests.set(key, entry);
    }
  };

  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;
    if (!(await readLines(file, openFile, onLine))) onError();
  }

  // Open days: every non-frozen day of the window, empty ones included, so
  // the caller can freeze them all (an unfrozen empty day would hold the
  // threshold back and every run would re-read the whole window).
  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 });
    if (!bucket.projects[r.project]) bucket.projects[r.project] = emptyProject();
    addProject(bucket.projects[r.project], {
      work: r.input + r.output + r.cacheCreation,
      cacheRead: r.cacheRead,
      requests: 1,
    });
  }

  // Merged 30-day series and project totals over the window.
  const days = [];
  const projectTotals = dict();
  for (const date of window) {
    const source = Object.prototype.hasOwnProperty.call(frozen, date)
      ? normalizeDay(frozen[date])
      : openDays[date];
    days.push({ date, models: capModels(source.models) });
    for (const [name, p] of Object.entries(source.projects)) {
      if (!projectTotals[name]) projectTotals[name] = emptyProject();
      addProject(projectTotals[name], p);
    }
  }

  return {
    openDays,
    days,
    projects: topProjects(projectTotals),
    claudeCodeVersion: latest ? latest.version : null,
    readErrors,
  };
}

// Top TOP_PROJECTS by work tokens over the window, then one "(autres)" entry
// aggregating the rest when it has any volume. The sum of `work` over the
// returned list always equals the window total.
function topProjects(totals) {
  const list = Object.entries(totals)
    .map(([name, p]) => ({ name, work: p.work, cacheRead: p.cacheRead, requests: p.requests }))
    .filter((p) => p.work > 0 || p.cacheRead > 0 || p.requests > 0)
    .sort(
      (a, b) =>
        b.work - a.work ||
        b.cacheRead - a.cacheRead ||
        (a.name < b.name ? -1 : a.name > b.name ? 1 : 0),
    );
  const top = list.slice(0, TOP_PROJECTS);
  const rest = list.slice(TOP_PROJECTS);
  if (rest.length) {
    const other = { name: OTHER_PROJECTS, work: 0, cacheRead: 0, requests: 0 };
    for (const p of rest) addProject(other, p);
    top.push(other);
  }
  return top;
}

module.exports = {
  aggregate,
  mapProject,
  torontoDay,
  shiftDay,
  windowDays,
  capModels,
  topProjects,
  listTranscripts,
  TIME_ZONE,
  WINDOW_DAYS,
  TOP_PROJECTS,
  MAX_MODELS_PER_DAY,
  OTHER_MODEL,
  OTHER_PROJECTS,
};