/** * Data collection, caching, and insights for the /usage dashboard. * * Performance model (see CHANGELOG 0.4.0): * - Session JSONL files are scanned at the buffer level. Only lines that can * possibly be a session header or an assistant message are UTF-8 decoded and * JSON.parsed — the multi-megabyte tool-result lines that dominate session * files are skipped with a cheap byte search. * - Per-file extraction results are persisted to an on-disk cache keyed by * (size, mtimeMs). Session files are append-only, so a warm load only * re-parses files that changed since the last run. */ import { readdir, readFile, rename, stat, writeFile } from "node:fs/promises"; import { homedir } from "node:os"; import { dirname, join } from "node:path"; // ============================================================================= // Types // ============================================================================= export interface TokenStats { total: number; input: number; output: number; cacheRead: number; cacheWrite: number; } export interface BaseStats { messages: number; cost: number; tokens: TokenStats; } export interface ModelStats extends BaseStats { sessions: Set; } export interface ProviderStats extends BaseStats { sessions: Set; models: Map; } export interface TotalStats extends BaseStats { sessions: number; } export interface Insight { /** Structure insights always show; alarms fire only when material. */ kind: "structure" | "alarm"; /** Leading stat, already formatted (e.g. "34%", "$446", "1.4×"). */ stat: string; headline: string; /** Dimmed follow-up line; empty string renders nothing. */ advice: string; } export interface PeriodInsights { insights: Insight[]; } interface CostCount { cost: number; messages: number; } interface PeriodRawData { totalCost: number; totalMessages: number; /** Messages at ≥ CTX_TAX_THRESHOLD context. */ ctxHigh: CostCount; /** Messages below CTX_LOW_THRESHOLD context (comparison group). */ ctxLow: CostCount; projectCosts: Map; sessionCosts: Map; /** Cost of each session's first-ever message falling in this period. */ upfrontCost: number; /** Cache misses after >TTL_GAP_MS idle — resuming after the cache expired. */ ttlMissCost: number; /** Cache misses right after a mid-session model switch (no idle gap). */ modelSwitchMissCost: number; /** Cache misses with no idle gap, compaction, or model switch — true prefix changes. */ prefixMissCost: number; reasoningTokens: number; outputTokens: number; cacheReadTokens: number; freshTokens: number; } /** Per-message adjacency info, computed on raw file order before dedupe. */ interface MessageMeta { /** Gap to the previous assistant message in the same file; -1 when unknown. */ gapMs: number; /** Context size of the previous assistant message in the same file; 0 when first. */ prevCtx: number; /** True when the provider/model differs from the previous assistant message. */ modelSwitched: boolean; /** True for the first deduped message of a session across all its files. */ isSessionStart: boolean; } export interface TrendInfo { /** Cost over the last 7 calendar days including today. */ last7Cost: number; /** Average weekly cost over the prior 28 days. */ priorWeeklyPace: number; } export interface TimeFilteredStats { providers: Map; totals: TotalStats; insights: PeriodInsights; } /** * One (provider, model, thinkingLevel) cell inside an hourly bucket. * Powers the graph explorer; built post-dedupe so it matches table totals. */ export interface HourlyCell { messages: number; cost: number; input: number; output: number; cacheRead: number; cacheWrite: number; reasoning: number; } /** Composite key: `${provider}\u0000${model}\u0000${thinkingLevel}` */ export type HourlyKey = string; export const HOURLY_KEY_SEP = "\u0000"; export function makeHourlyKey(provider: string, model: string, thinkingLevel: string): HourlyKey { return provider + HOURLY_KEY_SEP + model + HOURLY_KEY_SEP + thinkingLevel; } export function splitHourlyKey(key: HourlyKey): { provider: string; model: string; thinkingLevel: string } { const [provider = "", model = "", thinkingLevel = ""] = key.split(HOURLY_KEY_SEP); return { provider, model, thinkingLevel }; } export interface PeriodBounds { todayMs: number; weekStartMs: number; lastWeekStartMs: number; last30DaysStartMs: number; nowMs: number; } export interface UsageData { today: TimeFilteredStats; thisWeek: TimeFilteredStats; lastWeek: TimeFilteredStats; last30Days: TimeFilteredStats; allTime: TimeFilteredStats; /** Deduped usage bucketed by hour start (ms) → series key → metrics. */ hourly: Map>; bounds: PeriodBounds; } export type TabName = "today" | "thisWeek" | "lastWeek" | "last30Days" | "allTime"; export const TAB_ORDER: TabName[] = ["today", "thisWeek", "lastWeek", "last30Days", "allTime"]; export interface SessionMessage { provider: string; model: string; /** Thinking level active when the message was produced; "" when unknown. */ thinkingLevel: string; cost: number; input: number; output: number; cacheRead: number; cacheWrite: number; /** Reasoning output tokens reported by the provider (0 when absent). */ reasoning: number; timestamp: number; /** * True when a compaction entry occurred between the previous assistant * message and this one. Compaction legitimately changes the request prefix, * so such messages are excluded from prefix-change cache-miss accounting. */ afterCompaction: boolean; } export interface ParsedSessionFile { /** Empty string when the file has no session header — such files are ignored. */ sessionId: string; /** Working directory from the session header; "" when absent. */ cwd: string; /** Extracted assistant messages, pre-dedupe. Dedupe happens at aggregation. */ messages: SessionMessage[]; } // ============================================================================= // Paths // ============================================================================= export function getAgentDir(): string { // Replicate Pi's logic: respect PI_CODING_AGENT_DIR env var return process.env.PI_CODING_AGENT_DIR || join(homedir(), ".pi", "agent"); } export function getSessionsDir(): string { return join(getAgentDir(), "sessions"); } export function getDefaultCachePath(): string { return join(getAgentDir(), "usage-extension-cache.json"); } // ============================================================================= // Session file discovery // ============================================================================= async function collectSessionFilesRecursively(dir: string, files: string[], signal?: AbortSignal): Promise { try { const entries = await readdir(dir, { withFileTypes: true }); for (const entry of entries) { if (signal?.aborted) return; const entryPath = join(dir, entry.name); if (entry.isDirectory()) { await collectSessionFilesRecursively(entryPath, files, signal); } else if (entry.isFile() && entry.name.endsWith(".jsonl")) { files.push(entryPath); } } } catch { // Skip directories we can't read } } async function getAllSessionFiles(sessionsDir: string, signal?: AbortSignal): Promise { const files: string[] = []; await collectSessionFilesRecursively(sessionsDir, files, signal); files.sort(); return files; } // ============================================================================= // Session file parsing // ============================================================================= const NEWLINE = 0x0a; // Relevance patterns for the buffer-level pre-filter. Pi writes compact JSON // (`"role":"assistant"`), but imported/third-party session files have been seen // with Python-style spaced JSON (`"role": "assistant"`), so both are matched. // False positives (e.g. a tool result quoting one of these strings verbatim) // only cost a wasted JSON.parse — the parsed entry is still shape-checked. const PATTERN_ASSISTANT_COMPACT = Buffer.from('"role":"assistant"'); const PATTERN_ASSISTANT_SPACED = Buffer.from('"role": "assistant"'); const PATTERN_SESSION_COMPACT = Buffer.from('"type":"session"'); const PATTERN_SESSION_SPACED = Buffer.from('"type": "session"'); const PATTERN_THINKING_COMPACT = Buffer.from('"type":"thinking_level_change"'); const PATTERN_THINKING_SPACED = Buffer.from('"type": "thinking_level_change"'); const PATTERN_COMPACTION_COMPACT = Buffer.from('"type":"compaction"'); const PATTERN_COMPACTION_SPACED = Buffer.from('"type": "compaction"'); const PARSE_YIELD_EVERY_LINES = 2000; function lineMightBeRelevant(line: Buffer): boolean { return ( line.includes(PATTERN_ASSISTANT_COMPACT) || line.includes(PATTERN_SESSION_COMPACT) || line.includes(PATTERN_THINKING_COMPACT) || line.includes(PATTERN_COMPACTION_COMPACT) || line.includes(PATTERN_ASSISTANT_SPACED) || line.includes(PATTERN_SESSION_SPACED) || line.includes(PATTERN_THINKING_SPACED) || line.includes(PATTERN_COMPACTION_SPACED) ); } /** * Extract the session id and assistant messages from a session JSONL buffer. * Returns partial results when aborted — callers must check `signal.aborted` * before caching or using the result. */ export async function parseSessionBuffer(buffer: Buffer, signal?: AbortSignal): Promise { const messages: SessionMessage[] = []; let sessionId = ""; let cwd = ""; // Assistant messages don't carry the thinking level; pi records it as separate // thinking_level_change entries, always written before the first assistant // message of a session. Replaying them in append order attributes each message // to the level active when it was produced. let thinkingLevel = ""; let compactionPending = false; let start = 0; let lineNumber = 0; while (start < buffer.length) { let end = buffer.indexOf(NEWLINE, start); if (end === -1) end = buffer.length; lineNumber++; if (lineNumber % PARSE_YIELD_EVERY_LINES === 0) { await new Promise((resolve) => setImmediate(resolve)); if (signal?.aborted) return { sessionId, cwd, messages }; } if (end > start && lineMightBeRelevant(buffer.subarray(start, end))) { try { const entry = JSON.parse(buffer.toString("utf8", start, end)); if (entry.type === "session") { sessionId = entry.id; if (typeof entry.cwd === "string") cwd = entry.cwd; } else if (entry.type === "thinking_level_change") { if (typeof entry.thinkingLevel === "string") thinkingLevel = entry.thinkingLevel; } else if (entry.type === "compaction") { compactionPending = true; } else if (entry.type === "message" && entry.message?.role === "assistant") { const msg = entry.message; if (msg.usage && msg.provider && msg.model) { const fallbackTs = entry.timestamp ? new Date(entry.timestamp).getTime() : 0; messages.push({ provider: msg.provider, model: msg.model, thinkingLevel, cost: msg.usage.cost?.total || 0, input: msg.usage.input || 0, output: msg.usage.output || 0, cacheRead: msg.usage.cacheRead || 0, cacheWrite: msg.usage.cacheWrite || 0, reasoning: msg.usage.reasoning || 0, timestamp: msg.timestamp || (Number.isNaN(fallbackTs) ? 0 : fallbackTs), afterCompaction: compactionPending, }); compactionPending = false; } } } catch { // Skip malformed lines } } start = end + 1; } return { sessionId, cwd, messages }; } // ============================================================================= // On-disk cache // ============================================================================= const CACHE_VERSION = 3; type CachedMessageTuple = [ providerIdx: number, modelIdx: number, cost: number, input: number, output: number, cacheRead: number, cacheWrite: number, timestamp: number, thinkingLevelIdx: number, reasoning: number, afterCompaction: 0 | 1, ]; interface CacheFileEntry { size: number; mtimeMs: number; sessionId: string; cwd: string; messages: CachedMessageTuple[]; } export interface CachedFileState { size: number; mtimeMs: number; parsed: ParsedSessionFile; } export async function loadUsageCache(cachePath: string): Promise> { const result = new Map(); let raw: { version?: unknown; names?: unknown; files?: unknown }; try { raw = JSON.parse(await readFile(cachePath, "utf8")); } catch { return result; // Missing or corrupt cache — rebuild from scratch. } if (!raw || raw.version !== CACHE_VERSION || !Array.isArray(raw.names) || typeof raw.files !== "object" || raw.files === null) { return result; } const names = raw.names as unknown[]; for (const [filePath, entry] of Object.entries(raw.files as Record)) { if ( !entry || typeof entry.size !== "number" || typeof entry.mtimeMs !== "number" || typeof entry.sessionId !== "string" || typeof entry.cwd !== "string" || !Array.isArray(entry.messages) ) { continue; } const messages: SessionMessage[] = []; let valid = true; for (const tuple of entry.messages) { if (!Array.isArray(tuple) || tuple.length !== 11) { valid = false; break; } const provider = names[tuple[0]]; const model = names[tuple[1]]; const thinkingLevel = names[tuple[8]]; if (typeof provider !== "string" || typeof model !== "string" || typeof thinkingLevel !== "string") { valid = false; break; } messages.push({ provider, model, thinkingLevel, cost: Number(tuple[2]) || 0, input: Number(tuple[3]) || 0, output: Number(tuple[4]) || 0, cacheRead: Number(tuple[5]) || 0, cacheWrite: Number(tuple[6]) || 0, timestamp: Number(tuple[7]) || 0, reasoning: Number(tuple[9]) || 0, afterCompaction: tuple[10] === 1, }); } if (!valid) continue; result.set(filePath, { size: entry.size, mtimeMs: entry.mtimeMs, parsed: { sessionId: entry.sessionId, cwd: entry.cwd, messages }, }); } return result; } export async function saveUsageCache(cachePath: string, states: Map): Promise { const names: string[] = []; const nameIndex = new Map(); const intern = (name: string): number => { let idx = nameIndex.get(name); if (idx === undefined) { idx = names.length; names.push(name); nameIndex.set(name, idx); } return idx; }; const files: Record = {}; for (const [filePath, state] of states) { files[filePath] = { size: state.size, mtimeMs: state.mtimeMs, sessionId: state.parsed.sessionId, cwd: state.parsed.cwd, messages: state.parsed.messages.map((m): CachedMessageTuple => [ intern(m.provider), intern(m.model), m.cost, m.input, m.output, m.cacheRead, m.cacheWrite, m.timestamp, intern(m.thinkingLevel), m.reasoning, m.afterCompaction ? 1 : 0, ]), }; } const payload = JSON.stringify({ version: CACHE_VERSION, names, files }); // Atomic-ish write: concurrent /usage runs race to a last-writer-wins rename // instead of interleaving partial writes. const tmpPath = join(dirname(cachePath), `.usage-cache-${process.pid}-${Date.now()}.tmp`); await writeFile(tmpPath, payload, "utf8"); await rename(tmpPath, cachePath); } // ============================================================================= // Aggregation // ============================================================================= function emptyTokens(): TokenStats { return { total: 0, input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }; } function emptyModelStats(): ModelStats { return { sessions: new Set(), messages: 0, cost: 0, tokens: emptyTokens() }; } function emptyProviderStats(): ProviderStats { return { sessions: new Set(), messages: 0, cost: 0, tokens: emptyTokens(), models: new Map() }; } function emptyTimeFilteredStats(): TimeFilteredStats { return { providers: new Map(), totals: { sessions: 0, messages: 0, cost: 0, tokens: emptyTokens() }, insights: { insights: [] }, }; } function emptyPeriodRawData(): PeriodRawData { return { totalCost: 0, totalMessages: 0, ctxHigh: { cost: 0, messages: 0 }, ctxLow: { cost: 0, messages: 0 }, projectCosts: new Map(), sessionCosts: new Map(), upfrontCost: 0, ttlMissCost: 0, modelSwitchMissCost: 0, prefixMissCost: 0, reasoningTokens: 0, outputTokens: 0, cacheReadTokens: 0, freshTokens: 0, }; } /** * Collapse a session cwd to a short, stable project label: `~` for the home * directory, up to two path segments below home (worktrees collapse to their * repository), absolute paths elsewhere. */ export function projectLabelFromCwd(cwd: string): string { if (!cwd) return "(unknown)"; // Collapse any home-directory prefix to "~", not just the current user's — // session stores merged from other machines can carry a different username. const home = homedir(); let homePrefix: string | null = null; if (cwd === home || cwd.startsWith(home + "/")) { homePrefix = home; } else { const m = /^(\/Users\/[^/]+|\/home\/[^/]+)(?=\/|$)/.exec(cwd); if (m) homePrefix = m[1]!; } if (homePrefix !== null && cwd.length <= homePrefix.length) return "~"; let rel = homePrefix !== null ? cwd.slice(homePrefix.length + 1) : cwd; const wt = rel.indexOf("/.worktrees/"); if (wt !== -1) rel = rel.slice(0, wt); const parts = rel.split("/").filter(Boolean); const label = parts.slice(0, 2).join("/"); return homePrefix !== null ? `~/${label}` : `/${label}`; } function emptyUsageData(bounds: PeriodBounds): UsageData { return { today: emptyTimeFilteredStats(), thisWeek: emptyTimeFilteredStats(), lastWeek: emptyTimeFilteredStats(), last30Days: emptyTimeFilteredStats(), allTime: emptyTimeFilteredStats(), hourly: new Map(), bounds, }; } const HOUR_MS = 3_600_000; function addToHourlyBuckets(hourly: Map>, msg: SessionMessage): void { if (msg.timestamp <= 0) return; // Unknown time can't be placed on a time axis. const hour = Math.floor(msg.timestamp / HOUR_MS) * HOUR_MS; let bucket = hourly.get(hour); if (!bucket) { bucket = new Map(); hourly.set(hour, bucket); } const key = makeHourlyKey(msg.provider, msg.model, msg.thinkingLevel); let cell = bucket.get(key); if (!cell) { cell = { messages: 0, cost: 0, input: 0, output: 0, cacheRead: 0, cacheWrite: 0, reasoning: 0 }; bucket.set(key, cell); } cell.messages++; cell.cost += msg.cost; cell.input += msg.input; cell.output += msg.output; cell.cacheRead += msg.cacheRead; cell.cacheWrite += msg.cacheWrite; cell.reasoning += msg.reasoning; } // Helper to accumulate stats into a target function accumulateStats( target: BaseStats, cost: number, tokens: { total: number; input: number; output: number; cacheRead: number; cacheWrite: number } ): void { target.messages++; target.cost += cost; target.tokens.total += tokens.total; target.tokens.input += tokens.input; target.tokens.output += tokens.output; target.tokens.cacheRead += tokens.cacheRead; target.tokens.cacheWrite += tokens.cacheWrite; } function getPeriodsForTimestamp( timestamp: number, todayMs: number, weekStartMs: number, lastWeekStartMs: number, last30DaysStartMs: number ): TabName[] { const periods: TabName[] = ["allTime"]; if (timestamp >= todayMs) periods.push("today"); if (timestamp >= weekStartMs) { periods.push("thisWeek"); } else if (timestamp >= lastWeekStartMs) { periods.push("lastWeek"); } if (timestamp >= last30DaysStartMs) periods.push("last30Days"); return periods; } const DAY_MS = 24 * HOUR_MS; const PROGRESS_REPORT_EVERY = 100; function addMessagesToUsageData( data: UsageData, sessionId: string, project: string, messages: SessionMessage[], meta: MessageMeta[], todayMs: number, weekStartMs: number, lastWeekStartMs: number, last30DaysStartMs: number, rawByPeriod: Record, costByDayIdx: Map ): void { const sessionContributed = { today: false, thisWeek: false, lastWeek: false, last30Days: false, allTime: false }; for (let mi = 0; mi < messages.length; mi++) { const msg = messages[mi]!; const mm = meta[mi]!; // pi's built-in test providers never call a real API — keep them out of stats. if (EXCLUDED_PROVIDERS.has(msg.provider)) continue; // Day-indexed cost totals power the burn-trend insight. if (msg.timestamp > 0) { const dayIdx = Math.floor((msg.timestamp - todayMs) / DAY_MS); costByDayIdx.set(dayIdx, (costByDayIdx.get(dayIdx) ?? 0) + msg.cost); } addToHourlyBuckets(data.hourly, msg); const periods = getPeriodsForTimestamp(msg.timestamp, todayMs, weekStartMs, lastWeekStartMs, last30DaysStartMs); const tokens = { // Count fresh tokens processed this turn. // Include cacheWrite because those prompt tokens were newly written and billed. // Exclude cacheRead because repeated cache hits would otherwise dominate totals. total: msg.input + msg.output + msg.cacheWrite, input: msg.input, output: msg.output, cacheRead: msg.cacheRead, cacheWrite: msg.cacheWrite, }; for (const period of periods) { const stats = data[period]; let providerStats = stats.providers.get(msg.provider); if (!providerStats) { providerStats = emptyProviderStats(); stats.providers.set(msg.provider, providerStats); } let modelStats = providerStats.models.get(msg.model); if (!modelStats) { modelStats = emptyModelStats(); providerStats.models.set(msg.model, modelStats); } modelStats.sessions.add(sessionId); accumulateStats(modelStats, msg.cost, tokens); providerStats.sessions.add(sessionId); accumulateStats(providerStats, msg.cost, tokens); accumulateStats(stats.totals, msg.cost, tokens); sessionContributed[period] = true; const raw = rawByPeriod[period]; raw.totalCost += msg.cost; raw.totalMessages++; const ctx = msg.input + msg.cacheRead + msg.cacheWrite; if (ctx >= CTX_TAX_THRESHOLD) { raw.ctxHigh.cost += msg.cost; raw.ctxHigh.messages++; } else if (ctx < CTX_LOW_THRESHOLD) { raw.ctxLow.cost += msg.cost; raw.ctxLow.messages++; } raw.projectCosts.set(project, (raw.projectCosts.get(project) ?? 0) + msg.cost); raw.sessionCosts.set(sessionId, (raw.sessionCosts.get(sessionId) ?? 0) + msg.cost); if (mm.isSessionStart) raw.upfrontCost += msg.cost; if ( !msg.afterCompaction && mm.prevCtx >= MISS_MIN_PREV_CONTEXT && msg.cacheRead < Math.min(MISS_MAX_CACHE_READ, 0.3 * mm.prevCtx) ) { if (mm.gapMs > TTL_GAP_MS) raw.ttlMissCost += msg.cost; else if (mm.gapMs >= 0 && mm.modelSwitched) raw.modelSwitchMissCost += msg.cost; else if (mm.gapMs >= 0) raw.prefixMissCost += msg.cost; } raw.reasoningTokens += msg.reasoning; raw.outputTokens += msg.output; raw.cacheReadTokens += msg.cacheRead; raw.freshTokens += msg.input + msg.cacheWrite; } } if (sessionContributed.today) data.today.totals.sessions++; if (sessionContributed.thisWeek) data.thisWeek.totals.sessions++; if (sessionContributed.lastWeek) data.lastWeek.totals.sessions++; if (sessionContributed.last30Days) data.last30Days.totals.sessions++; if (sessionContributed.allTime) data.allTime.totals.sessions++; } // ============================================================================= // Collection orchestration // ============================================================================= const STAT_CONCURRENCY = 16; const DEFAULT_PARSE_CONCURRENCY = 4; const AGGREGATE_YIELD_EVERY_FILES = 200; export interface CollectProgress { /** Why this pass needs to parse files. */ mode: "first-run" | "rebuild" | "update"; /** Session files that need parsing this pass (0 = fully warm). */ filesToParse: number; /** Files parsed so far; reported in coarse increments. */ filesParsed: number; /** Newest session activity already ingested (ms since epoch); null when starting fresh. */ sinceMs: number | null; } export interface CollectUsageOptions { signal?: AbortSignal; /** Called once before parsing begins and periodically while files are parsed. */ onProgress?: (progress: CollectProgress) => void; /** Defaults to `/sessions`. */ sessionsDir?: string; /** Defaults to `/usage-extension-cache.json`. Pass `null` to disable the on-disk cache. */ cachePath?: string | null; /** Reference time for period bucketing. Defaults to `new Date()`. */ now?: Date; parseConcurrency?: number; } export async function collectUsageData(options: CollectUsageOptions = {}): Promise { const signal = options.signal; const now = options.now ?? new Date(); const sessionsDir = options.sessionsDir ?? getSessionsDir(); const cachePath = options.cachePath === undefined ? getDefaultCachePath() : options.cachePath; const parseConcurrency = Math.max(1, options.parseConcurrency ?? DEFAULT_PARSE_CONCURRENCY); const startOfToday = new Date(now); startOfToday.setHours(0, 0, 0, 0); const todayMs = startOfToday.getTime(); // Start of current week (Monday 00:00) const startOfWeek = new Date(now); const dayOfWeek = startOfWeek.getDay(); // 0 = Sunday, 1 = Monday, ... const daysSinceMonday = dayOfWeek === 0 ? 6 : dayOfWeek - 1; startOfWeek.setDate(startOfWeek.getDate() - daysSinceMonday); startOfWeek.setHours(0, 0, 0, 0); const weekStartMs = startOfWeek.getTime(); // Start of last week (previous Monday 00:00) const startOfLastWeek = new Date(startOfWeek); startOfLastWeek.setDate(startOfLastWeek.getDate() - 7); const lastWeekStartMs = startOfLastWeek.getTime(); // Rolling 30-day window: the last 30 calendar days including today, // i.e. from midnight 29 days before today. setDate handles DST correctly. const startOfLast30Days = new Date(startOfToday); startOfLast30Days.setDate(startOfLast30Days.getDate() - 29); const last30DaysStartMs = startOfLast30Days.getTime(); // 1. Discover session files. const filePaths = await getAllSessionFiles(sessionsDir, signal); if (signal?.aborted) return null; // 2. Stat them (batched) so cache freshness can be checked without reading contents. const fileStats = new Map(); { let next = 0; await Promise.all( Array.from({ length: STAT_CONCURRENCY }, async () => { while (next < filePaths.length) { if (signal?.aborted) return; const filePath = filePaths[next++]!; try { const st = await stat(filePath); fileStats.set(filePath, { size: st.size, mtimeMs: st.mtimeMs }); } catch { // File vanished between listing and stat — skip it. } } }) ); } if (signal?.aborted) return null; // 3. Load the cache and decide which files actually need parsing. let cacheFileExists = false; if (cachePath) { try { await stat(cachePath); cacheFileExists = true; } catch { // No cache file yet — first run. } } const previous = cachePath ? await loadUsageCache(cachePath) : new Map(); if (signal?.aborted) return null; const current = new Map(); const toParse: string[] = []; for (const filePath of filePaths) { const st = fileStats.get(filePath); if (!st) continue; const cached = previous.get(filePath); if (cached && cached.size === st.size && cached.mtimeMs === st.mtimeMs) { current.set(filePath, cached); } else { toParse.push(filePath); } } let dirty = toParse.length > 0; if (!dirty) { for (const filePath of previous.keys()) { if (!fileStats.has(filePath)) { dirty = true; // A cached file was deleted — evict it by rewriting. break; } } } // Progress reporting: distinguish a true first run from a format-change // rebuild (cache file present but unusable) and a routine incremental update. const progressMode: CollectProgress["mode"] = previous.size > 0 ? "update" : cacheFileExists ? "rebuild" : "first-run"; let sinceMs: number | null = null; if (progressMode === "update") { for (const state of previous.values()) { if (state.mtimeMs > (sinceMs ?? 0)) sinceMs = state.mtimeMs; } } let filesParsed = 0; const reportProgress = (): void => { options.onProgress?.({ mode: progressMode, filesToParse: toParse.length, filesParsed, sinceMs }); }; reportProgress(); // 4. Parse new/changed files with bounded concurrency. { let next = 0; await Promise.all( Array.from({ length: parseConcurrency }, async () => { while (next < toParse.length) { if (signal?.aborted) return; const filePath = toParse[next++]!; const st = fileStats.get(filePath)!; let buffer: Buffer; try { buffer = await readFile(filePath); } catch { filesParsed++; continue; // File vanished — skip it. } const parsed = await parseSessionBuffer(buffer, signal); if (signal?.aborted) return; // Never cache a partial parse. current.set(filePath, { size: st.size, mtimeMs: st.mtimeMs, parsed }); filesParsed++; if (filesParsed % PROGRESS_REPORT_EVERY === 0 || filesParsed === toParse.length) { reportProgress(); } } }) ); } if (signal?.aborted) { // Best effort: persist whatever finished so a cancelled cold build makes // the next attempt cheaper. Keep old entries for files not processed yet — // they are re-validated against size/mtime next run anyway. if (cachePath && dirty && current.size > 0) { const merged = new Map(previous); for (const [filePath, state] of current) merged.set(filePath, state); await saveUsageCache(cachePath, merged).catch(() => {}); } return null; } // 5. Persist the refreshed cache (also evicts entries for deleted files). if (cachePath && dirty) { await saveUsageCache(cachePath, current).catch(() => { // Cache write failures must never break /usage. }); } // 6. Aggregate in sorted path order with cross-file dedupe. const data = emptyUsageData({ todayMs, weekStartMs, lastWeekStartMs, last30DaysStartMs, nowMs: now.getTime() }); const rawByPeriod: Record = { today: emptyPeriodRawData(), thisWeek: emptyPeriodRawData(), lastWeek: emptyPeriodRawData(), last30Days: emptyPeriodRawData(), allTime: emptyPeriodRawData(), }; const costByDayIdx = new Map(); const seenSessions = new Set(); const seenHashes = new Set(); let processedFiles = 0; for (const filePath of filePaths) { const state = current.get(filePath); if (!state || !state.parsed.sessionId) continue; if (++processedFiles % AGGREGATE_YIELD_EVERY_FILES === 0) { await new Promise((resolve) => setImmediate(resolve)); if (signal?.aborted) return null; } // Deduplicate copied history across branched session files, computing // adjacency metadata (idle gaps, previous context) on the raw file order // so branch copies do not distort miss classification. const rawMsgs = state.parsed.messages; const deduped: SessionMessage[] = []; const meta: MessageMeta[] = []; for (let i = 0; i < rawMsgs.length; i++) { const m = rawMsgs[i]!; const hash = `${m.timestamp}:${m.input + m.output + m.cacheRead + m.cacheWrite}`; if (seenHashes.has(hash)) continue; seenHashes.add(hash); const prev = i > 0 ? rawMsgs[i - 1]! : null; deduped.push(m); meta.push({ gapMs: prev && prev.timestamp > 0 && m.timestamp > 0 ? m.timestamp - prev.timestamp : -1, prevCtx: prev ? prev.input + prev.cacheRead + prev.cacheWrite : 0, modelSwitched: prev !== null && (prev.provider !== m.provider || prev.model !== m.model), isSessionStart: false, }); } if (deduped.length === 0) continue; if (!seenSessions.has(state.parsed.sessionId)) { seenSessions.add(state.parsed.sessionId); meta[0]!.isSessionStart = true; } addMessagesToUsageData( data, state.parsed.sessionId, projectLabelFromCwd(state.parsed.cwd), deduped, meta, todayMs, weekStartMs, lastWeekStartMs, last30DaysStartMs, rawByPeriod, costByDayIdx ); } // Burn trend: last 7 calendar days vs the average weekly pace of the prior 28. let last7 = 0; let prior28 = 0; for (const [idx, c] of costByDayIdx) { if (idx >= -6) last7 += c; else if (idx >= -34) prior28 += c; } const trend: TrendInfo | null = prior28 > 0 ? { last7Cost: last7, priorWeeklyPace: prior28 / 4 } : null; for (const period of TAB_ORDER) { data[period].insights = computeInsights(rawByPeriod[period], trend); } return data; } // ============================================================================= // Insights // ============================================================================= // Context tax (structure) const CTX_TAX_THRESHOLD = 150_000; const CTX_LOW_THRESHOLD = 100_000; // Project mix (structure) const PROJECT_TOP_COUNT = 3; const PROJECT_MAX_DOMINANCE_PERCENT = 90; // Reasoning share (structure) const REASONING_MIN_PERCENT = 5; // Burn trend (structure) const TREND_HIGH_RATIO = 1.5; const TREND_LOW_RATIO = 0.6; // Cache-miss alarms const TTL_GAP_MS = 5 * 60_000; const MISS_MIN_PREV_CONTEXT = 20_000; const MISS_MAX_CACHE_READ = 5_000; /** pi's built-in test providers never send anything to a real API. */ const EXCLUDED_PROVIDERS = new Set(["faux-provider", "fake-provider"]); const CACHE_MISS_ALARM_PERCENT = 2; const CACHE_MISS_ALARM_MIN_COST = 1; // Concentration / upfront alarms const TOP_SESSION_COUNT = 5; const CONCENTRATION_ALARM_PERCENT = 35; const UPFRONT_ALARM_PERCENT = 8; // Cache-leverage alarm const LEVERAGE_FLOOR = 5; const LEVERAGE_MIN_COST = 5; const LEVERAGE_MIN_FRESH_TOKENS = 1_000_000; function fmtMoney(v: number): string { if (v >= 1000) return `$${(v / 1000).toFixed(1)}k`; if (v >= 100) return `$${Math.round(v)}`; return `$${v.toFixed(2)}`; } function fmtPercent(p: number): string { return p >= 10 ? `${Math.round(p)}%` : `${p.toFixed(1)}%`; } /** * Insights come in two kinds: * - structure: always-on decomposition of where the period's cost went. * - alarm: fires only when a wasteful pattern is material for the period, so * an all-clear period shows a calm panel instead of a wall of 2% factoids. * Periods with zero recorded cost produce an empty list — the UI renders a * distinct empty-state for that case. */ function computeInsights(raw: PeriodRawData, trend: TrendInfo | null): PeriodInsights { if (raw.totalMessages === 0 || raw.totalCost <= 0) { return { insights: [] }; } const total = raw.totalCost; const insights: Insight[] = []; // --- Alarms (listed first) --- const ttlPct = (raw.ttlMissCost / total) * 100; if (ttlPct >= CACHE_MISS_ALARM_PERCENT && raw.ttlMissCost >= CACHE_MISS_ALARM_MIN_COST) { insights.push({ kind: "alarm", stat: fmtMoney(raw.ttlMissCost), headline: `spent resuming conversations after a break (${fmtPercent(ttlPct)} of this period)`, advice: "Sent context is only reusable for a few minutes. After a longer pause, the next message pays to send the whole conversation again. Replying while a session is fresh avoids this.", }); } const switchPct = (raw.modelSwitchMissCost / total) * 100; if (switchPct >= CACHE_MISS_ALARM_PERCENT && raw.modelSwitchMissCost >= CACHE_MISS_ALARM_MIN_COST) { insights.push({ kind: "alarm", stat: fmtMoney(raw.modelSwitchMissCost), headline: `spent switching models mid-conversation (${fmtPercent(switchPct)} of this period)`, advice: "Changing model re-sends the whole conversation at full price — the previous model's saved context doesn't transfer. Switching between tasks instead of mid-conversation avoids this.", }); } const prefixPct = (raw.prefixMissCost / total) * 100; if (prefixPct >= CACHE_MISS_ALARM_PERCENT && raw.prefixMissCost >= CACHE_MISS_ALARM_MIN_COST) { insights.push({ kind: "alarm", stat: fmtMoney(raw.prefixMissCost), headline: `spent re-sending conversations mid-session (${fmtPercent(prefixPct)} of this period)`, advice: "These messages paid full price for context that had already been sent — with no break, compaction, or model switch to explain it. Usually a tool or workflow is restarting or rewriting conversations. Worth a look if it stays high.", }); } if (raw.sessionCosts.size > TOP_SESSION_COUNT) { const sortedSessions = Array.from(raw.sessionCosts.values()).sort((a, b) => b - a); const topWeight = sortedSessions.slice(0, TOP_SESSION_COUNT).reduce((sum, c) => sum + c, 0); const topPct = (topWeight / total) * 100; if (topPct >= CONCENTRATION_ALARM_PERCENT) { insights.push({ kind: "alarm", stat: fmtMoney(topWeight), headline: `came from just ${TOP_SESSION_COUNT} of your ${raw.sessionCosts.size} sessions (${fmtPercent(topPct)} of this period)`, advice: "A handful of sessions drove most of the spend. The graph view can show what they were doing.", }); } } const upfrontPct = (raw.upfrontCost / total) * 100; if (upfrontPct >= UPFRONT_ALARM_PERCENT) { insights.push({ kind: "alarm", stat: fmtMoney(raw.upfrontCost), headline: `spent on the opening message of new sessions (${fmtPercent(upfrontPct)} of this period)`, advice: "A session's first message sends everything from scratch. Fewer, longer sessions cut this overhead.", }); } if (total >= LEVERAGE_MIN_COST && raw.freshTokens >= LEVERAGE_MIN_FRESH_TOKENS) { const leverage = raw.cacheReadTokens / raw.freshTokens; if (leverage < LEVERAGE_FLOOR) { insights.push({ kind: "alarm", stat: `${leverage.toFixed(1)}×`, headline: "tokens reused from history for every token paid at full price", advice: "Typical interactive use reuses 10× or more. A low number means conversations keep being sent from scratch — look for workflows that restart sessions.", }); } } // --- Structure (always-on) --- if (raw.ctxHigh.messages > 0) { const pct = (raw.ctxHigh.cost / total) * 100; if (pct >= 1) { const avgHigh = raw.ctxHigh.cost / raw.ctxHigh.messages; const avgLow = raw.ctxLow.messages > 0 ? raw.ctxLow.cost / raw.ctxLow.messages : 0; const cmp = avgLow > 0 ? ` — ${fmtMoney(avgHigh)}/msg vs ${fmtMoney(avgLow)} under ${formatThresholdTokens(CTX_LOW_THRESHOLD)}` : ""; insights.push({ kind: "structure", stat: fmtPercent(pct), headline: `of your cost came from messages with ≥${formatThresholdTokens(CTX_TAX_THRESHOLD)} tokens loaded${cmp}`, advice: "Long conversations cost more per message. /compact mid-task and /clear between tasks keep them lean.", }); } } if (raw.projectCosts.size >= 2) { const top = [...raw.projectCosts.entries()].sort((a, b) => b[1] - a[1]).slice(0, PROJECT_TOP_COUNT); const topPct = (top[0]![1] / total) * 100; if (topPct < PROJECT_MAX_DOMINANCE_PERCENT) { const rest = top .slice(1) .map(([label, c]) => `${label} ${fmtPercent((c / total) * 100)}`) .join(", "); insights.push({ kind: "structure", stat: fmtPercent(topPct), headline: `of your cost was ${top[0]![0]}${rest ? ` — then ${rest}` : ""}`, advice: "", }); } } if (raw.outputTokens > 0) { const reasoningPct = (raw.reasoningTokens / raw.outputTokens) * 100; if (reasoningPct >= REASONING_MIN_PERCENT) { insights.push({ kind: "structure", stat: fmtPercent(reasoningPct), headline: "of your output tokens were hidden reasoning", advice: "Models charge for their behind-the-scenes thinking as output tokens. pi records this only from 0.80.3 (June 2026), so older periods understate it.", }); } } if (trend && trend.priorWeeklyPace > 0) { const ratio = trend.last7Cost / trend.priorWeeklyPace; const advice = ratio >= TREND_HIGH_RATIO ? "Spending is up against your own baseline — the graph view shows what changed." : ratio <= TREND_LOW_RATIO ? "Spending is well below your recent baseline." : ""; insights.push({ kind: "structure", stat: `${ratio.toFixed(1)}×`, headline: `your last 7 days (${fmtMoney(trend.last7Cost)}) vs your prior 4-week pace (${fmtMoney(trend.priorWeeklyPace)}/wk)`, advice, }); } return { insights }; } function formatThresholdTokens(n: number): string { if (n >= 1_000_000) return `${n / 1_000_000}M`; if (n >= 1_000) return `${n / 1_000}k`; return String(n); }