feat(stats): add pure token-usage parser + cache-aware cost derivation

lib/stats/token-usage.mjs

@@ -0,0 +1,161 @@
+// lib/stats/token-usage.mjs
+// SKAL-2 — pure token-usage parser + cache-aware cost derivation for Voyage
+// observability. Captures MAIN-CONTEXT token/cost from the Claude Code
+// transcript (transcript_path). Each assistant record carries message.usage
+// with a per-REQUEST snapshot.
+//
+// Verified 2026-06-26 against a real local transcript (216 lines): 70
+// assistant records collapse to 31 distinct requestIds — records duplicate
+// per requestId (streamed snapshots), so we dedup by requestId keeping the
+// LAST occurrence (GH #28197 streaming-placeholder mitigation), then sum
+// across requests. The main transcript carried zero isSidechain:true records.
+//
+// v1 scope = MAIN-CONTEXT only. Sub-agent (swarm) turns live in separate
+// agent-*.jsonl siblings and are a documented v2 follow-on — so the
+// main-transcript sum UNDER-counts total Voyage cost. Every record is stamped
+// scope:'main-context' so no reader mistakes cost_usd for the session total.
+//
+// All functions in this module are PURE (no I/O). The impure capture shell
+// (captureTokenUsage) is added in Step 4.
+//
+// Zero npm dependencies. Node stdlib only.
+
+// Per-Mtok USD prices, resolved 2026-06-26 via the claude-api skill reference:
+// base input/output from the model table; cache rates from the prompt-caching
+// doc multipliers (cache_read 0.1x, write_5m 1.25x, write_1h 2.0x of input).
+export const PRICE_TABLE = Object.freeze({
+  'claude-opus-4-8': Object.freeze({
+    input: 5.0,
+    output: 25.0,
+    cache_read: 0.5,
+    cache_write_5m: 6.25,
+    cache_write_1h: 10.0,
+  }),
+});
+
+// Date the PRICE_TABLE values were resolved/verified. Bump when prices change.
+export const PRICE_TABLE_VERSION = '2026-06-26';
+
+function num(v) {
+  return typeof v === 'number' && Number.isFinite(v) ? v : 0;
+}
+
+/**
+ * Single pass over transcript JSONL text. Keeps only main-chain assistant
+ * records (type==='assistant', isSidechain!==true), dedups by requestId
+ * keeping the LAST occurrence (per-request usage snapshots duplicate across
+ * streamed records), and sums the four usage token fields. Malformed lines
+ * are skipped.
+ *
+ * @returns {{tokens_input:number, tokens_output:number,
+ *            tokens_cache_creation:number, tokens_cache_read:number}}
+ */
+export function parseTranscriptUsage(text) {
+  const lines = (text || '').split('\n');
+  // requestId -> usage (last occurrence wins). Records without a requestId
+  // get a unique sentinel key so each is still counted exactly once.
+  const byRequest = new Map();
+  let anon = 0;
+  for (const line of lines) {
+    const trimmed = line.trim();
+    if (trimmed === '') continue;
+    let obj;
+    try { obj = JSON.parse(trimmed); }
+    catch { continue; }
+    if (!obj || obj.type !== 'assistant') continue;
+    if (obj.isSidechain === true) continue;
+    const usage = obj.message && obj.message.usage;
+    if (!usage || typeof usage !== 'object') continue;
+    const key = typeof obj.requestId === 'string' && obj.requestId
+      ? obj.requestId
+      : `__anon_${anon++}`;
+    byRequest.set(key, usage); // last occurrence wins
+  }
+  const totals = {
+    tokens_input: 0,
+    tokens_output: 0,
+    tokens_cache_creation: 0,
+    tokens_cache_read: 0,
+  };
+  for (const usage of byRequest.values()) {
+    totals.tokens_input += num(usage.input_tokens);
+    totals.tokens_output += num(usage.output_tokens);
+    totals.tokens_cache_creation += num(usage.cache_creation_input_tokens);
+    totals.tokens_cache_read += num(usage.cache_read_input_tokens);
+  }
+  return totals;
+}
+
+/**
+ * Cache-aware USD cost. Lumped cache_creation is priced at the 5m write rate
+ * (Claude Code's default cache TTL; verified the dominant case on a real
+ * transcript — the 1h split is a documented v2 refinement). When the model is
+ * absent from the table we REFUSE to estimate: {cost_usd:null, is_estimate:true}.
+ *
+ * @returns {{cost_usd:number|null, is_estimate:boolean}}
+ */
+export function deriveCost(totals, model, priceTable = PRICE_TABLE) {
+  const price = priceTable && priceTable[model];
+  if (!price) return { cost_usd: null, is_estimate: true };
+  const t = totals || {};
+  const cost =
+    (num(t.tokens_input) * price.input +
+      num(t.tokens_output) * price.output +
+      num(t.tokens_cache_creation) * price.cache_write_5m +
+      num(t.tokens_cache_read) * price.cache_read) / 1_000_000;
+  return { cost_usd: cost, is_estimate: false };
+}
+
+/**
+ * Build the flat numeric record. Stamps scope:'main-context' and
+ * price_table_version. session_id is written for upsert keying but MUST be
+ * stripped at export (CWE-212, enforced by the field allowlist in Step 2).
+ *
+ * @param {{sessionId:string, model:string, totals:object,
+ *          priceTable?:object, now:string}} args
+ */
+export function buildRecord({ sessionId, model, totals, priceTable = PRICE_TABLE, now }) {
+  const t = totals || {};
+  const { cost_usd, is_estimate } = deriveCost(t, model, priceTable);
+  return {
+    ts: now,
+    session_id: sessionId,
+    scope: 'main-context',
+    model,
+    tokens_input: num(t.tokens_input),
+    tokens_output: num(t.tokens_output),
+    tokens_cache_creation: num(t.tokens_cache_creation),
+    tokens_cache_read: num(t.tokens_cache_read),
+    cost_usd,
+    is_estimate,
+    price_table_version: PRICE_TABLE_VERSION,
+  };
+}
+
+/**
+ * Pure read-modify-write. Parses existing JSONL lines, REPLACES the line whose
+ * session_id matches the record (dropping any further same-session dupes),
+ * else APPENDS. Returns the new file text (one line per session, trailing
+ * newline). One-line-per-session is what keeps cross-session sums from
+ * double-counting (vs append-and-grow).
+ */
+export function upsertSessionRecord(existingText, record) {
+  const lines = (existingText || '').split('\n');
+  const out = [];
+  let replaced = false;
+  for (const line of lines) {
+    const trimmed = line.trim();
+    if (trimmed === '') continue;
+    let obj;
+    try { obj = JSON.parse(trimmed); }
+    catch { out.push(trimmed); continue; } // preserve unparseable lines verbatim
+    if (obj && obj.session_id === record.session_id) {
+      if (!replaced) { out.push(JSON.stringify(record)); replaced = true; }
+      // else: duplicate same-session line — drop it
+    } else {
+      out.push(trimmed);
+    }
+  }
+  if (!replaced) out.push(JSON.stringify(record));
+  return out.join('\n') + '\n';
+}

tests/fixtures/token-usage/sample-transcript.jsonl

@@ -0,0 +1,8 @@
+{"type":"user","message":{"role":"user","content":"hi"}}
+{"type":"assistant","isSidechain":false,"requestId":"req_A","message":{"model":"claude-opus-4-8","usage":{"input_tokens":1,"output_tokens":1,"cache_creation_input_tokens":0,"cache_read_input_tokens":0}}}
+{"type":"assistant","isSidechain":false,"requestId":"req_A","message":{"model":"claude-opus-4-8","usage":{"input_tokens":1000,"output_tokens":200,"cache_creation_input_tokens":400,"cache_read_input_tokens":5000}}}
+{"type":"assistant","isSidechain":false,"requestId":"req_B","message":{"model":"claude-opus-4-8","usage":{"input_tokens":2000,"output_tokens":300,"cache_creation_input_tokens":600,"cache_read_input_tokens":1000}}}
+{"type":"assistant","isSidechain":true,"requestId":"req_SIDE","message":{"model":"claude-opus-4-8","usage":{"input_tokens":9999,"output_tokens":9999,"cache_creation_input_tokens":9999,"cache_read_input_tokens":9999}}}
+{"type":"system","content":"some system note with no usage"}
+this is not valid json and must be skipped
+{"type":"assistant","isSidechain":false,"requestId":"req_B","message":{"model":"claude-opus-4-8","usage":{"input_tokens":2000,"output_tokens":300,"cache_creation_input_tokens":600,"cache_read_input_tokens":1000}}}

tests/lib/token-usage.test.mjs

@@ -0,0 +1,142 @@
+// tests/lib/token-usage.test.mjs
+// SKAL-2 Step 1 — pure token-usage parser + cache-aware cost derivation.
+//
+// Hermetic: no LLM, network, time, or randomness. Fixture transcript exercises
+// the dedup-by-requestId (streaming-placeholder) and isSidechain-exclusion
+// paths that were verified against a real local transcript at execute time.
+// Idiom: tests/lib/coordinator-contract.test.mjs (pure-module node:test style).
+
+import { test } from 'node:test';
+import { strict as assert } from 'node:assert';
+import { readFileSync } from 'node:fs';
+import { join, dirname } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import {
+  parseTranscriptUsage,
+  deriveCost,
+  buildRecord,
+  upsertSessionRecord,
+  PRICE_TABLE,
+  PRICE_TABLE_VERSION,
+} from '../../lib/stats/token-usage.mjs';
+
+const HERE = dirname(fileURLToPath(import.meta.url));
+const FIXTURE = join(HERE, '..', 'fixtures', 'token-usage', 'sample-transcript.jsonl');
+
+// ---- parseTranscriptUsage --------------------------------------------------
+
+test('parseTranscriptUsage — sums main-chain final-per-requestId; excludes sidechain + placeholder dup', () => {
+  const text = readFileSync(FIXTURE, 'utf-8');
+  const totals = parseTranscriptUsage(text);
+  // req_A: placeholder {1,1,0,0} then real {1000,200,400,5000} → LAST wins.
+  // req_B: appears twice identically → dedup → counted once {2000,300,600,1000}.
+  // req_SIDE (isSidechain:true, all 9999) → EXCLUDED. user/system/malformed → skipped.
+  assert.deepEqual(totals, {
+    tokens_input: 3000,            // 1000 + 2000  (NOT 1 — placeholder dropped)
+    tokens_output: 500,            // 200 + 300
+    tokens_cache_creation: 1000,   // 400 + 600
+    tokens_cache_read: 6000,       // 5000 + 1000
+  });
+  // Sidechain 9999s must not leak into any field.
+  assert.ok(totals.tokens_input < 9999, 'sidechain tokens leaked into total');
+});
+
+test('parseTranscriptUsage — empty / nullish input → zeroed totals', () => {
+  const zero = { tokens_input: 0, tokens_output: 0, tokens_cache_creation: 0, tokens_cache_read: 0 };
+  assert.deepEqual(parseTranscriptUsage(''), zero);
+  assert.deepEqual(parseTranscriptUsage(null), zero);
+  assert.deepEqual(parseTranscriptUsage(undefined), zero);
+});
+
+test('parseTranscriptUsage — single-pass O(n) shape on a 500-line transcript', () => {
+  // 500 distinct-requestId main-chain assistant records, each input_tokens:10.
+  const lines = [];
+  for (let i = 0; i < 500; i++) {
+    lines.push(JSON.stringify({
+      type: 'assistant',
+      isSidechain: false,
+      requestId: `req_${i}`,
+      message: { model: 'claude-opus-4-8', usage: { input_tokens: 10, output_tokens: 0, cache_creation_input_tokens: 0, cache_read_input_tokens: 0 } },
+    }));
+  }
+  const totals = parseTranscriptUsage(lines.join('\n'));
+  assert.equal(totals.tokens_input, 5000); // 500 × 10
+});
+
+// ---- deriveCost ------------------------------------------------------------
+
+test('deriveCost — hand-computed value for a known model (cache-aware)', () => {
+  const totals = { tokens_input: 3000, tokens_output: 500, tokens_cache_creation: 1000, tokens_cache_read: 6000 };
+  const { cost_usd, is_estimate } = deriveCost(totals, 'claude-opus-4-8');
+  // (3000×5 + 500×25 + 1000×6.25 + 6000×0.5) / 1e6
+  // = (15000 + 12500 + 6250 + 3000) / 1e6 = 36750 / 1e6 = 0.03675
+  assert.equal(cost_usd, 0.03675);
+  assert.equal(is_estimate, false);
+});
+
+test('deriveCost — refuse-to-estimate for an unknown model', () => {
+  const totals = { tokens_input: 3000, tokens_output: 500, tokens_cache_creation: 1000, tokens_cache_read: 6000 };
+  const out = deriveCost(totals, 'claude-unknown-9');
+  assert.equal(out.cost_usd, null);
+  assert.equal(out.is_estimate, true);
+});
+
+test('PRICE_TABLE — frozen, opus-4-8 carries all five rate fields', () => {
+  assert.ok(Object.isFrozen(PRICE_TABLE), 'PRICE_TABLE must be frozen (drift-pin)');
+  const p = PRICE_TABLE['claude-opus-4-8'];
+  for (const k of ['input', 'output', 'cache_read', 'cache_write_5m', 'cache_write_1h']) {
+    assert.equal(typeof p[k], 'number', `missing rate field ${k}`);
+  }
+});
+
+// ---- buildRecord -----------------------------------------------------------
+
+test('buildRecord — stamps scope:main-context + price_table_version + flat numerics', () => {
+  const totals = { tokens_input: 3000, tokens_output: 500, tokens_cache_creation: 1000, tokens_cache_read: 6000 };
+  const rec = buildRecord({ sessionId: 'S1', model: 'claude-opus-4-8', totals, now: '2026-06-26T12:00:00Z' });
+  assert.equal(rec.scope, 'main-context');
+  assert.equal(rec.price_table_version, PRICE_TABLE_VERSION);
+  assert.equal(rec.ts, '2026-06-26T12:00:00Z');
+  assert.equal(rec.session_id, 'S1');
+  assert.equal(rec.model, 'claude-opus-4-8');
+  assert.equal(rec.tokens_input, 3000);
+  assert.equal(rec.cost_usd, 0.03675);
+  assert.equal(rec.is_estimate, false);
+});
+
+test('buildRecord — unknown model → cost_usd null + is_estimate true (honesty contract)', () => {
+  const rec = buildRecord({ sessionId: 'S9', model: 'mystery', totals: { tokens_input: 1 }, now: 't' });
+  assert.equal(rec.cost_usd, null);
+  assert.equal(rec.is_estimate, true);
+});
+
+// ---- upsertSessionRecord ---------------------------------------------------
+
+test('upsertSessionRecord — REPLACES a same-session line (not append)', () => {
+  const existing =
+    JSON.stringify({ session_id: 'S1', tokens_input: 1, scope: 'main-context' }) + '\n' +
+    JSON.stringify({ session_id: 'S2', tokens_input: 2, scope: 'main-context' }) + '\n';
+  const rec = buildRecord({ sessionId: 'S1', model: 'claude-opus-4-8', totals: { tokens_input: 999 }, now: 't' });
+  const out = upsertSessionRecord(existing, rec);
+  const recs = out.trim().split('\n').map(JSON.parse);
+  assert.equal(recs.length, 2, 'replace must NOT add a line');
+  const s1 = recs.find(r => r.session_id === 'S1');
+  assert.equal(s1.tokens_input, 999, 'S1 must carry the new totals');
+  assert.ok(recs.some(r => r.session_id === 'S2'), 'S2 must be preserved');
+});
+
+test('upsertSessionRecord — APPENDS a new-session line', () => {
+  const existing = JSON.stringify({ session_id: 'S1', tokens_input: 1 }) + '\n';
+  const rec = buildRecord({ sessionId: 'S3', model: 'claude-opus-4-8', totals: { tokens_input: 7 }, now: 't' });
+  const out = upsertSessionRecord(existing, rec);
+  const recs = out.trim().split('\n').map(JSON.parse);
+  assert.equal(recs.length, 2, 'new session must append');
+  assert.ok(recs.some(r => r.session_id === 'S3' && r.tokens_input === 7));
+});
+
+test('upsertSessionRecord — empty file → single record, trailing newline', () => {
+  const rec = buildRecord({ sessionId: 'S1', model: 'claude-opus-4-8', totals: { tokens_input: 5 }, now: 't' });
+  const out = upsertSessionRecord('', rec);
+  assert.ok(out.endsWith('\n'));
+  assert.deepEqual(out.trim().split('\n').map(JSON.parse).length, 1);
+});