From 3109a3b5a4b76deda496caae5d61d87cbbd81fc8 Mon Sep 17 00:00:00 2001 From: teamchong <25894545+teamchong@users.noreply.github.com> Date: Mon, 18 May 2026 16:13:04 -0400 Subject: [PATCH] cli: add 'pixelpipe stats' subcommand for offline log analysis MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reads ~/.pixelpipe/events.jsonl (or --file / PIXELPIPE_LOG) and prints a one-screen summary of how the proxy is doing: - request counts by status bucket (2xx/4xx/5xx) - compression rate (compressed vs passthrough) - latency p50/p95/p99 (duration + first-byte separately) - aggregate Anthropic token usage (input/output/cache_create/cache_read) - cache hit rate computed two ways (by tokens, by events) - top skip reasons (so we see WHY a request wasn't compressed) - top cwds with per-project compression ratio - system_sha8 distribution → 'reuse rate' = how often the cached payload is being re-sent. High reuse + low cache_read means something is breaking cache_control. - unknown_static_tags warning section if Claude Code shipped a new dynamic tag that we're silently baking into the image. --json mode dumps the raw Summary as JSON for downstream pipelines. Streaming reader so a 100 MB log file doesn't OOM. Pure aggregator (fold/newSummary/renderTextReport) is unit-tested; only the file IO shell is untested. Dispatched from src/node.ts BEFORE parseCli so its flags don't collide with the proxy's flags. The CLI shim in bin/cli.js stays unchanged. Tests: +6 (status bucketing, compressed/passthrough split, token agg, cwd & sha histogram, unknown-tag collection, text report rendering). --- src/node.ts | 14 +- src/stats.ts | 341 ++++++++++++++++++++++++++++++++++++++++++++ tests/stats.test.ts | 136 ++++++++++++++++++ 3 files changed, 489 insertions(+), 2 deletions(-) create mode 100644 src/stats.ts create mode 100644 tests/stats.test.ts diff --git a/src/node.ts b/src/node.ts index 8a85946..7a4f5d5 100644 --- a/src/node.ts +++ b/src/node.ts @@ -82,7 +82,8 @@ function printHelp(): void { console.log(`pixelpipe — token-saving proxy for Claude Code Usage: - pixelpipe [options] + pixelpipe [options] run the proxy + pixelpipe stats [--file

] summarize ~/.pixelpipe/events.jsonl Options: -p, --port listen port (default 47821) @@ -288,7 +289,16 @@ class FileTracker implements Tracker { // ---- main ---------------------------------------------------------------- async function main(): Promise { - const opts = parseCli(process.argv.slice(2)); + // Subcommand dispatch — before parseCli so flag parsing doesn't choke on + // `stats`'s own flags. + const argv = process.argv.slice(2); + if (argv[0] === 'stats') { + const { runStats } = await import('./stats.js'); + const code = await runStats(argv.slice(1)); + process.exit(code); + } + + const opts = parseCli(argv); const transform: TransformOptions = { compress: opts.compress, compressTools: opts.compressTools, diff --git a/src/stats.ts b/src/stats.ts new file mode 100644 index 0000000..dffb252 --- /dev/null +++ b/src/stats.ts @@ -0,0 +1,341 @@ +/** + * `pixelpipe stats` — read the JSONL events file (or any file matching the + * tracker.ts schema) and print aggregate metrics about how the proxy is + * doing. + * + * Node-only (uses node:fs). Streams the file line-by-line so a 100 MB log + * doesn't blow the heap. The aggregator itself is pure — fed a sequence of + * TrackEvent and produces a Summary — so a Workers-side dashboard could + * reuse it later by extracting it into core/. + * + * Exit codes: + * 0 ok, summary printed + * 1 events file missing or unreadable + * 2 events file exists but contained zero valid lines + */ + +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; +import * as readline from 'node:readline'; +import type { TrackEvent } from './core/tracker.js'; + +// ---- pure aggregator ------------------------------------------------------ + +export interface Summary { + total: number; + ok2xx: number; + err4xx: number; + err5xx: number; + compressed: number; + passthrough: number; + /** Sum of orig_chars across compressed requests — the bytes we removed + * from the text path by rendering to PNG. */ + origCharsTotal: number; + imageBytesTotal: number; + /** Aggregated Anthropic token usage. */ + inputTokensTotal: number; + outputTokensTotal: number; + cacheCreateTokensTotal: number; + cacheReadTokensTotal: number; + /** Number of events whose cache_read_tokens > 0 — i.e. the prompt cache + * actually hit. */ + cacheHitEvents: number; + /** Number of events that carried any usage data at all. Denominator for + * cacheHitEvents. */ + eventsWithUsage: number; + durationMs: number[]; + firstByteMs: number[]; + skipReasons: Map; + byCwd: Map; + /** system_sha8 → number of times seen. High repeat count = cache should + * be doing its job. */ + systemShaHist: Map; + unknownTags: Map; +} + +export function newSummary(): Summary { + return { + total: 0, + ok2xx: 0, + err4xx: 0, + err5xx: 0, + compressed: 0, + passthrough: 0, + origCharsTotal: 0, + imageBytesTotal: 0, + inputTokensTotal: 0, + outputTokensTotal: 0, + cacheCreateTokensTotal: 0, + cacheReadTokensTotal: 0, + cacheHitEvents: 0, + eventsWithUsage: 0, + durationMs: [], + firstByteMs: [], + skipReasons: new Map(), + byCwd: new Map(), + systemShaHist: new Map(), + unknownTags: new Map(), + }; +} + +export function fold(s: Summary, ev: TrackEvent): Summary { + s.total++; + if (ev.status >= 200 && ev.status < 300) s.ok2xx++; + else if (ev.status >= 400 && ev.status < 500) s.err4xx++; + else if (ev.status >= 500) s.err5xx++; + + if (ev.compressed === true) { + s.compressed++; + if (typeof ev.orig_chars === 'number') s.origCharsTotal += ev.orig_chars; + if (typeof ev.image_bytes === 'number') s.imageBytesTotal += ev.image_bytes; + } else if (ev.compressed === false) { + s.passthrough++; + if (ev.reason) s.skipReasons.set(ev.reason, (s.skipReasons.get(ev.reason) ?? 0) + 1); + } + + if (typeof ev.duration_ms === 'number') s.durationMs.push(ev.duration_ms); + if (typeof ev.first_byte_ms === 'number') s.firstByteMs.push(ev.first_byte_ms); + + const hasUsage = + typeof ev.input_tokens === 'number' || + typeof ev.cache_read_tokens === 'number' || + typeof ev.cache_create_tokens === 'number' || + typeof ev.output_tokens === 'number'; + if (hasUsage) { + s.eventsWithUsage++; + s.inputTokensTotal += ev.input_tokens ?? 0; + s.outputTokensTotal += ev.output_tokens ?? 0; + s.cacheCreateTokensTotal += ev.cache_create_tokens ?? 0; + s.cacheReadTokensTotal += ev.cache_read_tokens ?? 0; + if ((ev.cache_read_tokens ?? 0) > 0) s.cacheHitEvents++; + } + + if (ev.cwd) { + const k = ev.cwd; + const e = s.byCwd.get(k) ?? { count: 0, origChars: 0, imageBytes: 0 }; + e.count++; + e.origChars += ev.orig_chars ?? 0; + e.imageBytes += ev.image_bytes ?? 0; + s.byCwd.set(k, e); + } + + if (ev.system_sha8) { + s.systemShaHist.set(ev.system_sha8, (s.systemShaHist.get(ev.system_sha8) ?? 0) + 1); + } + + if (ev.unknown_static_tags) { + for (const t of ev.unknown_static_tags) { + s.unknownTags.set(t, (s.unknownTags.get(t) ?? 0) + 1); + } + } + + return s; +} + +function percentile(sorted: number[], p: number): number { + if (sorted.length === 0) return 0; + const idx = Math.min(sorted.length - 1, Math.floor((p / 100) * sorted.length)); + return sorted[idx]!; +} + +/** Format a number with thousands separators. Used for big token counts. */ +function fmtN(n: number): string { + return n.toLocaleString('en-US'); +} + +function fmtPct(num: number, denom: number): string { + if (denom === 0) return ' —'; + return ((num / denom) * 100).toFixed(1).padStart(4) + '%'; +} + +// ---- text report ---------------------------------------------------------- + +export function renderTextReport(s: Summary): string { + const lines: string[] = []; + const sortedDur = [...s.durationMs].sort((a, b) => a - b); + const sortedFB = [...s.firstByteMs].sort((a, b) => a - b); + + lines.push('━━━ pixelpipe stats ━━━'); + lines.push(''); + lines.push(`requests: ${fmtN(s.total)}`); + lines.push( + ` 2xx: ${fmtN(s.ok2xx).padStart(8)} ` + + `4xx: ${fmtN(s.err4xx).padStart(6)} 5xx: ${fmtN(s.err5xx).padStart(6)}`, + ); + lines.push( + ` compressed: ${fmtN(s.compressed).padStart(8)} (${fmtPct(s.compressed, s.total)})`, + ); + lines.push( + ` passthrough: ${fmtN(s.passthrough).padStart(8)} (${fmtPct(s.passthrough, s.total)})`, + ); + lines.push(''); + + lines.push('latency (ms):'); + lines.push( + ` duration p50=${percentile(sortedDur, 50)} p95=${percentile(sortedDur, 95)} p99=${percentile(sortedDur, 99)}`, + ); + lines.push( + ` first-byte p50=${percentile(sortedFB, 50)} p95=${percentile(sortedFB, 95)} p99=${percentile(sortedFB, 99)}`, + ); + lines.push(''); + + lines.push('compression:'); + lines.push(` orig text rendered: ${fmtN(s.origCharsTotal)} chars`); + lines.push(` image bytes: ${fmtN(s.imageBytesTotal)} B`); + const ratio = + s.origCharsTotal > 0 ? (s.imageBytesTotal / s.origCharsTotal).toFixed(3) : '—'; + lines.push(` bytes/char ratio: ${ratio}`); + lines.push(''); + + lines.push('Anthropic token usage:'); + lines.push(` input: ${fmtN(s.inputTokensTotal).padStart(12)}`); + lines.push(` output: ${fmtN(s.outputTokensTotal).padStart(12)}`); + lines.push(` cache create: ${fmtN(s.cacheCreateTokensTotal).padStart(12)}`); + lines.push(` cache read: ${fmtN(s.cacheReadTokensTotal).padStart(12)}`); + const totalIn = + s.inputTokensTotal + s.cacheCreateTokensTotal + s.cacheReadTokensTotal; + lines.push( + ` cache hit rate (by tokens): ${fmtPct(s.cacheReadTokensTotal, totalIn)}`, + ); + lines.push( + ` cache hit rate (by events): ${fmtPct(s.cacheHitEvents, s.eventsWithUsage)}`, + ); + lines.push(''); + + if (s.skipReasons.size > 0) { + lines.push('top skip reasons:'); + const top = [...s.skipReasons.entries()].sort((a, b) => b[1] - a[1]).slice(0, 5); + for (const [reason, count] of top) { + lines.push(` ${count.toString().padStart(6)} ${reason}`); + } + lines.push(''); + } + + if (s.byCwd.size > 0) { + lines.push('top working dirs (by request count):'); + const top = [...s.byCwd.entries()].sort((a, b) => b[1].count - a[1].count).slice(0, 10); + for (const [cwd, e] of top) { + const cratio = e.origChars > 0 ? (e.imageBytes / e.origChars).toFixed(2) : '—'; + lines.push(` ${e.count.toString().padStart(6)} ratio=${cratio} ${cwd}`); + } + lines.push(''); + } + + if (s.systemShaHist.size > 0) { + lines.push('top system prompts (system_sha8, high count = cache reuse):'); + const top = [...s.systemShaHist.entries()].sort((a, b) => b[1] - a[1]).slice(0, 5); + for (const [sha, count] of top) { + lines.push(` ${count.toString().padStart(6)} ${sha}`); + } + const unique = s.systemShaHist.size; + const reuseRate = + s.total > 0 ? (((s.total - unique) / s.total) * 100).toFixed(1) : '—'; + lines.push(` unique prompts: ${unique} reuse rate: ${reuseRate}%`); + lines.push(''); + } + + if (s.unknownTags.size > 0) { + lines.push('⚠ unknown tag-shaped blocks observed in static slab:'); + const top = [...s.unknownTags.entries()].sort((a, b) => b[1] - a[1]); + for (const [tag, count] of top) { + lines.push(` ${count.toString().padStart(6)} <${tag}>`); + } + lines.push( + ' → consider adding these to DYNAMIC_BLOCK_TAGS in src/core/transform.ts', + ); + lines.push(''); + } + + return lines.join('\n'); +} + +// ---- entrypoint ----------------------------------------------------------- + +interface StatsOpts { + file: string; + json: boolean; +} + +function parseArgs(argv: string[]): StatsOpts { + const opts: StatsOpts = { + file: + process.env.PIXELPIPE_LOG ?? + path.join(os.homedir(), '.pixelpipe', 'events.jsonl'), + json: false, + }; + for (let i = 0; i < argv.length; i++) { + const a = argv[i]!; + if (a === '--file' || a === '-f') opts.file = argv[++i] ?? opts.file; + else if (a === '--json') opts.json = true; + else if (a === '-h' || a === '--help') { + console.log(`pixelpipe stats — aggregate metrics from events JSONL + +Usage: + pixelpipe stats [--file ] [--json] + +Options: + -f, --file events JSONL to read (default ~/.pixelpipe/events.jsonl + or PIXELPIPE_LOG env var) + --json emit the Summary object as JSON instead of a text report + -h, --help show this help +`); + process.exit(0); + } + } + return opts; +} + +export async function runStats(argv: string[]): Promise { + const opts = parseArgs(argv); + + if (!fs.existsSync(opts.file)) { + console.error(`[pixelpipe stats] events file not found: ${opts.file}`); + console.error( + `[pixelpipe stats] (run pixelpipe and send a request first, or set PIXELPIPE_LOG)`, + ); + return 1; + } + + const stream = fs.createReadStream(opts.file, { encoding: 'utf8' }); + const rl = readline.createInterface({ input: stream, crlfDelay: Infinity }); + const summary = newSummary(); + let parsed = 0; + let dropped = 0; + for await (const line of rl) { + if (!line.trim()) continue; + try { + const ev = JSON.parse(line) as TrackEvent; + fold(summary, ev); + parsed++; + } catch { + dropped++; + } + } + + if (parsed === 0) { + console.error(`[pixelpipe stats] no valid events in ${opts.file}`); + return 2; + } + + if (opts.json) { + // Maps don't survive JSON.stringify — convert to plain objects. + process.stdout.write( + JSON.stringify( + { + ...summary, + skipReasons: Object.fromEntries(summary.skipReasons), + byCwd: Object.fromEntries(summary.byCwd), + systemShaHist: Object.fromEntries(summary.systemShaHist), + unknownTags: Object.fromEntries(summary.unknownTags), + }, + null, + 2, + ) + '\n', + ); + } else { + process.stdout.write(renderTextReport(summary) + '\n'); + if (dropped > 0) console.error(`(${dropped} unparseable line(s) skipped)`); + } + return 0; +} diff --git a/tests/stats.test.ts b/tests/stats.test.ts new file mode 100644 index 0000000..c2b8ffc --- /dev/null +++ b/tests/stats.test.ts @@ -0,0 +1,136 @@ +import { describe, it, expect } from 'vitest'; +import { newSummary, fold, renderTextReport } from '../src/stats.js'; +import type { TrackEvent } from '../src/core/tracker.js'; + +function ev(partial: Partial): TrackEvent { + return { + ts: '2026-05-18T00:00:00Z', + method: 'POST', + path: '/v1/messages', + status: 200, + duration_ms: 100, + ...partial, + }; +} + +describe('stats aggregator', () => { + it('counts status buckets', () => { + const s = newSummary(); + fold(s, ev({ status: 200 })); + fold(s, ev({ status: 201 })); + fold(s, ev({ status: 404 })); + fold(s, ev({ status: 503 })); + fold(s, ev({ status: 500 })); + expect(s.total).toBe(5); + expect(s.ok2xx).toBe(2); + expect(s.err4xx).toBe(1); + expect(s.err5xx).toBe(2); + }); + + it('separates compressed vs passthrough and collects skip reasons', () => { + const s = newSummary(); + fold(s, ev({ compressed: true, orig_chars: 1000, image_bytes: 200 })); + fold(s, ev({ compressed: true, orig_chars: 2000, image_bytes: 300 })); + fold(s, ev({ compressed: false, reason: 'below_min_chars (50 < 2000)' })); + fold(s, ev({ compressed: false, reason: 'below_min_chars (60 < 2000)' })); + fold(s, ev({ compressed: false, reason: 'compress=false' })); + expect(s.compressed).toBe(2); + expect(s.passthrough).toBe(3); + expect(s.origCharsTotal).toBe(3000); + expect(s.imageBytesTotal).toBe(500); + // Reasons keep their exact string form (parenthetical char counts and + // all) — useful for spotting outliers without collapsing detail. + expect(s.skipReasons.size).toBe(3); + expect(s.skipReasons.get('below_min_chars (50 < 2000)')).toBe(1); + expect(s.skipReasons.get('below_min_chars (60 < 2000)')).toBe(1); + expect(s.skipReasons.get('compress=false')).toBe(1); + }); + + it('aggregates Anthropic token usage and computes cache hit metrics', () => { + const s = newSummary(); + fold( + s, + ev({ + input_tokens: 100, + output_tokens: 10, + cache_read_tokens: 0, + cache_create_tokens: 5000, + }), + ); + fold( + s, + ev({ + input_tokens: 50, + output_tokens: 5, + cache_read_tokens: 5000, + cache_create_tokens: 0, + }), + ); + fold( + s, + ev({ + input_tokens: 60, + output_tokens: 6, + cache_read_tokens: 5000, + cache_create_tokens: 0, + }), + ); + // 3 events all carried usage; 2 had cache_read > 0. + expect(s.eventsWithUsage).toBe(3); + expect(s.cacheHitEvents).toBe(2); + expect(s.inputTokensTotal).toBe(210); + expect(s.outputTokensTotal).toBe(21); + expect(s.cacheReadTokensTotal).toBe(10000); + expect(s.cacheCreateTokensTotal).toBe(5000); + }); + + it('buckets by cwd and tracks system_sha8 reuse', () => { + const s = newSummary(); + fold(s, ev({ cwd: '/a', system_sha8: 'aaa', orig_chars: 100, image_bytes: 20 })); + fold(s, ev({ cwd: '/a', system_sha8: 'aaa', orig_chars: 100, image_bytes: 20 })); + fold(s, ev({ cwd: '/b', system_sha8: 'bbb', orig_chars: 200, image_bytes: 40 })); + expect(s.byCwd.size).toBe(2); + expect(s.byCwd.get('/a')!.count).toBe(2); + expect(s.byCwd.get('/a')!.origChars).toBe(200); + expect(s.systemShaHist.get('aaa')).toBe(2); + expect(s.systemShaHist.get('bbb')).toBe(1); + }); + + it('collects unknown_static_tags across events', () => { + const s = newSummary(); + fold(s, ev({ unknown_static_tags: ['recent_files', 'todo_list'] })); + fold(s, ev({ unknown_static_tags: ['recent_files'] })); + fold(s, ev({})); + expect(s.unknownTags.get('recent_files')).toBe(2); + expect(s.unknownTags.get('todo_list')).toBe(1); + }); + + it('renders a non-empty text report for a populated summary', () => { + const s = newSummary(); + for (let i = 0; i < 100; i++) { + fold( + s, + ev({ + compressed: true, + orig_chars: 5000, + image_bytes: 1000, + input_tokens: 50, + cache_read_tokens: i % 2 === 0 ? 4000 : 0, + cache_create_tokens: i % 2 === 0 ? 0 : 4000, + duration_ms: 100 + i, + first_byte_ms: 30 + i, + cwd: '/Users/x/code/pp', + system_sha8: 'stable', + }), + ); + } + const out = renderTextReport(s); + expect(out).toContain('pixelpipe stats'); + expect(out).toContain('compressed'); + expect(out).toContain('cache hit rate'); + expect(out).toContain('/Users/x/code/pp'); + expect(out).toContain('stable'); + // 50% cache hit rate by event. + expect(out).toMatch(/cache hit rate \(by events\):\s+50.0%/); + }); +});