From a94cbf49663452d2ce202cb06874435aa8c45357 Mon Sep 17 00:00:00 2001 From: "Lim, Un-tiong" Date: Wed, 12 Aug 2026 20:02:42 +0800 Subject: [PATCH 1/4] fix: stream JSONL reads so token widgets work on large transcripts Claude Code session transcripts can exceed Node's max string length (~512MB). readJsonlLines used fs.readFile(..., 'utf-8'), which throws Cannot create a string longer than 0x1fffffe8 characters. getTokenMetrics swallowed that error and returned zeros, so In/Out/Total stayed at 0 while session-cost (from statusline stdin) still looked correct. Stream line-by-line via createReadStream/readline (async) and chunked Buffer reads (sync), and keep getTokenMetrics on the streaming path. Fixes #550 --- src/utils/__tests__/jsonl-lines.test.ts | 136 ++++++++++++++++++++++ src/utils/__tests__/jsonl-metrics.test.ts | 38 ++++++ src/utils/jsonl-lines.ts | 95 +++++++++++++-- src/utils/jsonl-metrics.ts | 48 ++++---- 4 files changed, 281 insertions(+), 36 deletions(-) create mode 100644 src/utils/__tests__/jsonl-lines.test.ts diff --git a/src/utils/__tests__/jsonl-lines.test.ts b/src/utils/__tests__/jsonl-lines.test.ts new file mode 100644 index 00000000..85fec379 --- /dev/null +++ b/src/utils/__tests__/jsonl-lines.test.ts @@ -0,0 +1,136 @@ +import * as fs from 'fs'; +import os from 'os'; +import path from 'path'; +import { + afterEach, + describe, + expect, + it +} from 'vitest'; + +import { + iterateJsonlLines, + iterateJsonlLinesSync, + readJsonlLines, + readJsonlLinesSync +} from '../jsonl-lines'; + +describe('jsonl line streaming', () => { + const tempRoots: string[] = []; + + afterEach(() => { + while (tempRoots.length > 0) { + const root = tempRoots.pop(); + if (root) { + fs.rmSync(root, { recursive: true, force: true }); + } + } + }); + + function writeTranscript(name: string, content: string): string { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-lines-')); + tempRoots.push(root); + const filePath = path.join(root, name); + fs.writeFileSync(filePath, content); + return filePath; + } + + it('reads lf and crlf lines without requiring a trailing newline', async () => { + const filePath = writeTranscript('mixed.jsonl', [ + '{"id":1}', + '{"id":2}\r', + '{"id":3}' + ].join('\n')); + + await expect(readJsonlLines(filePath)).resolves.toEqual([ + '{"id":1}', + '{"id":2}', + '{"id":3}' + ]); + expect(readJsonlLinesSync(filePath)).toEqual([ + '{"id":1}', + '{"id":2}', + '{"id":3}' + ]); + }); + + it('skips empty lines like the previous whole-file trim/split path', async () => { + const filePath = writeTranscript('empty-lines.jsonl', '\n{"a":1}\n\n{"b":2}\n\n'); + + await expect(readJsonlLines(filePath)).resolves.toEqual([ + '{"a":1}', + '{"b":2}' + ]); + expect(readJsonlLinesSync(filePath)).toEqual([ + '{"a":1}', + '{"b":2}' + ]); + }); + + it('handles multi-byte utf-8 sequences that span sync read chunks', () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-lines-')); + tempRoots.push(root); + const filePath = path.join(root, 'utf8.jsonl'); + + // Force the sync reader across many tiny chunks by writing a long prefix + // so the multi-byte character is unlikely to land on a single natural boundary + // only — the reader itself uses 1MB chunks; put the character mid-file. + const emoji = '😀'; // 4-byte UTF-8 + const prefix = `{"n":"${'x'.repeat(100)}"}`; + const mid = `{"emoji":"${emoji}"}`; + const suffix = `{"n":"${'y'.repeat(100)}"}`; + fs.writeFileSync(filePath, [prefix, mid, suffix].join('\n'), 'utf8'); + + const lines = readJsonlLinesSync(filePath); + expect(lines).toHaveLength(3); + expect(JSON.parse(lines[1]!).emoji).toBe(emoji); + }); + + it('streams via async iterator without loading the full file as one string', async () => { + const filePath = writeTranscript('stream.jsonl', [ + '{"line":1}', + '{"line":2}', + '{"line":3}' + ].join('\n')); + + const seen: string[] = []; + for await (const line of iterateJsonlLines(filePath)) { + seen.push(line); + } + expect(seen).toEqual([ + '{"line":1}', + '{"line":2}', + '{"line":3}' + ]); + + expect(Array.from(iterateJsonlLinesSync(filePath))).toEqual(seen); + }); + + it('can read files larger than Node max string length via streaming', async () => { + // Node refuses to create a single string longer than ~0x1fffffe8 (~512MB). + // Building a real 512MB+ fixture is too heavy for unit tests, so we prove + // the streaming path never calls readFile/readFileSync for the payload and + // still aggregates many chunks correctly by reading a multi-chunk file. + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-lines-')); + tempRoots.push(root); + const filePath = path.join(root, 'chunked.jsonl'); + + const lineCount = 5000; + const handle = fs.openSync(filePath, 'w'); + try { + for (let i = 0; i < lineCount; i++) { + fs.writeSync(handle, `{"i":${i},"pad":"${'z'.repeat(200)}"}\n`); + } + } finally { + fs.closeSync(handle); + } + + const lines = await readJsonlLines(filePath); + expect(lines).toHaveLength(lineCount); + expect(JSON.parse(lines[0]!).i).toBe(0); + expect(JSON.parse(lines[lineCount - 1]!).i).toBe(lineCount - 1); + + const syncLines = readJsonlLinesSync(filePath); + expect(syncLines).toHaveLength(lineCount); + }, 30000); +}); diff --git a/src/utils/__tests__/jsonl-metrics.test.ts b/src/utils/__tests__/jsonl-metrics.test.ts index d1d5e5e7..860d1173 100644 --- a/src/utils/__tests__/jsonl-metrics.test.ts +++ b/src/utils/__tests__/jsonl-metrics.test.ts @@ -513,6 +513,44 @@ describe('jsonl transcript metrics', () => { }); }); + it('aggregates token metrics by streaming many usage lines without a whole-file string read', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); + tempRoots.push(root); + const transcriptPath = path.join(root, 'streamed-tokens.jsonl'); + + // Many small lines so the streamer crosses chunk boundaries while the + // cumulative totals stay easy to assert. Regression for #550: full-file + // utf-8 reads throw once transcripts exceed Node's max string length. + const lineCount = 2500; + const handle = fs.openSync(transcriptPath, 'w'); + try { + for (let i = 0; i < lineCount; i++) { + fs.writeSync(handle, `${makeUsageLine({ + timestamp: `2026-01-01T10:${String(Math.floor(i / 60)).padStart(2, '0')}:${String(i % 60).padStart(2, '0')}.000Z`, + input: 2, + output: 3, + cacheRead: 4, + cacheCreate: 1, + stopReason: 'end_turn' + })}\n`); + } + } finally { + fs.closeSync(handle); + } + + const metrics = await getTokenMetrics(transcriptPath); + + expect(metrics).toEqual({ + inputTokens: lineCount * 2, + outputTokens: lineCount * 3, + cachedTokens: lineCount * 5, + cacheReadTokens: lineCount * 4, + cacheCreationTokens: lineCount * 1, + totalTokens: lineCount * 10, + contextLength: 2 + 4 + 1 + }); + }, 30000); + it('calculates speed metrics from user-to-assistant processing windows', async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-speed-')); tempRoots.push(root); diff --git a/src/utils/jsonl-lines.ts b/src/utils/jsonl-lines.ts index 3663a800..5b54a42f 100644 --- a/src/utils/jsonl-lines.ts +++ b/src/utils/jsonl-lines.ts @@ -1,21 +1,98 @@ import * as fs from 'fs'; -import { promisify } from 'util'; +import { createInterface } from 'node:readline'; -const readFile = promisify(fs.readFile); -const readFileSync = fs.readFileSync; +const SYNC_READ_CHUNK_BYTES = 1024 * 1024; -function splitJsonlContent(content: string): string[] { - return content.trim().split('\n').filter(line => line.length > 0); +/** + * Stream a JSONL file line-by-line without materializing the whole file as one + * string. Claude Code session transcripts can exceed Node's max string length + * (~512MB / 0x1fffffe8), so `fs.readFile(..., 'utf-8')` throws and callers that + * catch the error end up reporting zeros. + */ +export async function* iterateJsonlLines(filePath: string): AsyncGenerator { + const stream = fs.createReadStream(filePath, { encoding: 'utf8' }); + const reader = createInterface({ + input: stream, + crlfDelay: Infinity + }); + + try { + for await (const line of reader) { + if (line.length > 0) { + yield line; + } + } + } finally { + reader.close(); + stream.destroy(); + } +} + +/** + * Synchronous line iterator for call sites that cannot be async. + * Completes each line in a Buffer before decoding so multi-byte UTF-8 sequences + * are never split across chunk boundaries. + */ +export function* iterateJsonlLinesSync(filePath: string): Generator { + const fd = fs.openSync(filePath, 'r'); + try { + const scratch = Buffer.allocUnsafe(SYNC_READ_CHUNK_BYTES); + let pending = Buffer.alloc(0); + + for (;;) { + const bytesRead = fs.readSync(fd, scratch, 0, scratch.length, null); + if (bytesRead === 0) { + break; + } + + const chunk = scratch.subarray(0, bytesRead); + const combined = pending.length > 0 ? Buffer.concat([pending, chunk]) : chunk; + let start = 0; + + for (let i = 0; i < combined.length; i++) { + if (combined[i] !== 0x0a) { + continue; + } + + let lineBuf = combined.subarray(start, i); + if (lineBuf.length > 0 && lineBuf[lineBuf.length - 1] === 0x0d) { + lineBuf = lineBuf.subarray(0, lineBuf.length - 1); + } + if (lineBuf.length > 0) { + yield lineBuf.toString('utf8'); + } + start = i + 1; + } + + pending = start === 0 + ? Buffer.from(combined) + : Buffer.from(combined.subarray(start)); + } + + if (pending.length > 0) { + let lineBuf = pending; + if (lineBuf[lineBuf.length - 1] === 0x0d) { + lineBuf = lineBuf.subarray(0, lineBuf.length - 1); + } + if (lineBuf.length > 0) { + yield lineBuf.toString('utf8'); + } + } + } finally { + fs.closeSync(fd); + } } export async function readJsonlLines(filePath: string): Promise { - const content = await readFile(filePath, 'utf-8'); - return splitJsonlContent(content); + const lines: string[] = []; + for await (const line of iterateJsonlLines(filePath)) { + lines.push(line); + } + return lines; } export function readJsonlLinesSync(filePath: string): string[] { - const content = readFileSync(filePath, 'utf-8'); - return splitJsonlContent(content); + return Array.from(iterateJsonlLinesSync(filePath)); } export function parseJsonlLine(line: string): unknown { diff --git a/src/utils/jsonl-metrics.ts b/src/utils/jsonl-metrics.ts index 2321bf41..e233e24b 100644 --- a/src/utils/jsonl-metrics.ts +++ b/src/utils/jsonl-metrics.ts @@ -12,6 +12,7 @@ import { isCompactBoundary } from './compaction'; import { + iterateJsonlLines, parseJsonlLine, readJsonlLines } from './jsonl-lines'; @@ -91,39 +92,29 @@ export async function getSessionDuration(transcriptPath: string): Promise= 0; i--) { - const line = lines[i]; - if (!line) { + const timestamp = new Date(data.timestamp); + if (Number.isNaN(timestamp.getTime())) { continue; } - const data = parseJsonlLine(line) as { timestamp?: string } | null; - if (data?.timestamp) { - lastTimestamp = new Date(data.timestamp); - break; + if (!firstTimestamp) { + firstTimestamp = timestamp; } + lastTimestamp = timestamp; } - if (!firstTimestamp || !lastTimestamp) { + if (!sawAnyLine || !firstTimestamp || !lastTimestamp) { return null; } @@ -154,13 +145,12 @@ export async function getSessionDuration(transcriptPath: string): Promise { try { - // Use Node.js-compatible file reading + // Stream line-by-line. Full-file readFile('utf-8') throws once a session + // transcript exceeds Node's max string length (~512MB). if (!fs.existsSync(transcriptPath)) { return { inputTokens: 0, outputTokens: 0, cachedTokens: 0, cacheReadTokens: 0, cacheCreationTokens: 0, totalTokens: 0, contextLength: 0 }; } - const lines = await readJsonlLines(transcriptPath); - let inputTokens = 0; let outputTokens = 0; let cacheReadTokens = 0; @@ -187,10 +177,13 @@ export async function getTokenMetrics(transcriptPath: string): Promise lastCompactBoundaryLineIndex + if (entryLineIndex > lastCompactBoundaryLineIndex && (!mostRecentPostCompactionTimestamp || entryTime > mostRecentPostCompactionTimestamp)) { mostRecentPostCompactionTimestamp = entryTime; mostRecentPostCompactionEntry = data; From 7e7ec5542508e15f5f39d32464fc51f0e99c5ff7 Mon Sep 17 00:00:00 2001 From: Matthew Breedlove Date: Wed, 2 Sep 2026 23:20:16 -0400 Subject: [PATCH 2/4] fix: bound streaming transcript memory --- src/utils/__tests__/jsonl-lines.test.ts | 13 ++ src/utils/__tests__/jsonl-metrics.test.ts | 46 ++++++ src/utils/jsonl-lines.ts | 42 +++-- src/utils/jsonl-metrics.ts | 180 +++++++++++++++------- 4 files changed, 209 insertions(+), 72 deletions(-) diff --git a/src/utils/__tests__/jsonl-lines.test.ts b/src/utils/__tests__/jsonl-lines.test.ts index 2f9bbe27..a380d152 100644 --- a/src/utils/__tests__/jsonl-lines.test.ts +++ b/src/utils/__tests__/jsonl-lines.test.ts @@ -84,6 +84,19 @@ describe('jsonl line streaming', () => { expect(lines).toEqual([line]); }); + it('reads a record spanning many sync chunks followed by another record', () => { + const filePath = writeTranscript('long-record.jsonl', [ + `{"value":"${'x'.repeat(6 * 1024 * 1024)}"}`, + '{"value":"next"}' + ].join('\n')); + + const lines = Array.from(iterateJsonlLinesSync(filePath)); + + expect(lines).toHaveLength(2); + expect(nth(lines, 0)).toHaveLength((6 * 1024 * 1024) + 12); + expect(nth(lines, 1)).toBe('{"value":"next"}'); + }); + it('streams via async iterator without loading the full file as one string', async () => { const filePath = writeTranscript('stream.jsonl', [ '{"line":1}', diff --git a/src/utils/__tests__/jsonl-metrics.test.ts b/src/utils/__tests__/jsonl-metrics.test.ts index 860d1173..344761ea 100644 --- a/src/utils/__tests__/jsonl-metrics.test.ts +++ b/src/utils/__tests__/jsonl-metrics.test.ts @@ -551,6 +551,52 @@ describe('jsonl transcript metrics', () => { }); }, 30000); + it('discards large assistant content after extracting token fields', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); + tempRoots.push(root); + const transcriptPath = path.join(root, 'large-assistant-content.jsonl'); + const content = 'x'.repeat(2 * 1024 * 1024); + + fs.writeFileSync(transcriptPath, [ + JSON.stringify({ + timestamp: '2026-01-01T10:00:00.000Z', + message: { + content, + stop_reason: 'end_turn', + usage: { + input_tokens: 2, + output_tokens: 3, + cache_read_input_tokens: 4, + cache_creation_input_tokens: 1 + } + } + }), + JSON.stringify({ + timestamp: '2026-01-01T10:00:01.000Z', + message: { + content, + stop_reason: 'end_turn', + usage: { + input_tokens: 5, + output_tokens: 6, + cache_read_input_tokens: 7, + cache_creation_input_tokens: 2 + } + } + }) + ].join('\n')); + + await expect(getTokenMetrics(transcriptPath)).resolves.toEqual({ + inputTokens: 7, + outputTokens: 9, + cachedTokens: 14, + cacheReadTokens: 11, + cacheCreationTokens: 3, + totalTokens: 30, + contextLength: 14 + }); + }); + it('calculates speed metrics from user-to-assistant processing windows', async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-speed-')); tempRoots.push(root); diff --git a/src/utils/jsonl-lines.ts b/src/utils/jsonl-lines.ts index 76bd2fc3..f19de223 100644 --- a/src/utils/jsonl-lines.ts +++ b/src/utils/jsonl-lines.ts @@ -96,14 +96,15 @@ export async function* iterateJsonlLines(filePath: string): AsyncGenerator { const fd = fs.openSync(filePath, 'r'); try { const scratch = Buffer.allocUnsafe(SYNC_READ_CHUNK_BYTES); - let pending = Buffer.alloc(0); + const pending: Buffer[] = []; + let pendingBytes = 0; for (;;) { const bytesRead = fs.readSync(fd, scratch, 0, scratch.length, null); @@ -112,31 +113,46 @@ export function* iterateJsonlLinesSync(filePath: string): Generator { } const chunk = scratch.subarray(0, bytesRead); - const combined = pending.length > 0 ? Buffer.concat([pending, chunk]) : chunk; let start = 0; - for (let i = 0; i < combined.length; i++) { - if (combined[i] !== 0x0a) { + for (let i = 0; i < chunk.length; i++) { + if (chunk[i] !== 0x0a) { continue; } - let lineBuf = combined.subarray(start, i); + const segment = chunk.subarray(start, i); + let lineBuf: Buffer; + if (pending.length === 0) { + lineBuf = segment; + } else { + if (segment.length > 0) { + pending.push(segment); + pendingBytes += segment.length; + } + lineBuf = Buffer.concat(pending, pendingBytes); + } + + pending.length = 0; + pendingBytes = 0; + start = i + 1; + if (lineBuf.length > 0 && lineBuf[lineBuf.length - 1] === 0x0d) { lineBuf = lineBuf.subarray(0, lineBuf.length - 1); } if (lineBuf.length > 0) { yield lineBuf.toString('utf8'); } - start = i + 1; } - pending = start === 0 - ? Buffer.from(combined) - : Buffer.from(combined.subarray(start)); + if (start < chunk.length) { + const remainder = Buffer.from(chunk.subarray(start)); + pending.push(remainder); + pendingBytes += remainder.length; + } } - if (pending.length > 0) { - let lineBuf = pending; + if (pendingBytes > 0) { + let lineBuf = Buffer.concat(pending, pendingBytes); if (lineBuf[lineBuf.length - 1] === 0x0d) { lineBuf = lineBuf.subarray(0, lineBuf.length - 1); } diff --git a/src/utils/jsonl-metrics.ts b/src/utils/jsonl-metrics.ts index f5f5f2c8..705914ee 100644 --- a/src/utils/jsonl-metrics.ts +++ b/src/utils/jsonl-metrics.ts @@ -49,6 +49,77 @@ interface CollectedSpeedMetrics { latestTimestampMs: number | null; } +interface RetainedTokenUsage { + inputTokens: number; + outputTokens: number; + cacheReadTokens: number; + cacheCreationTokens: number; +} + +interface TokenMetricEntry { + usage: RetainedTokenUsage; + stopReason: string | null | undefined; + timestamp: string | undefined; + isMainChain: boolean; + lineIndex: number; +} + +interface TokenMetricAccumulator { + inputTokens: number; + outputTokens: number; + cacheReadTokens: number; + cacheCreationTokens: number; + mostRecentMainChainUsage: RetainedTokenUsage | null; + mostRecentTimestamp: Date | null; + mostRecentPostCompactionUsage: RetainedTokenUsage | null; + mostRecentPostCompactionTimestamp: Date | null; +} + +function createTokenMetricAccumulator(): TokenMetricAccumulator { + return { + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheCreationTokens: 0, + mostRecentMainChainUsage: null, + mostRecentTimestamp: null, + mostRecentPostCompactionUsage: null, + mostRecentPostCompactionTimestamp: null + }; +} + +function resetPostCompactionUsage(accumulator: TokenMetricAccumulator): void { + accumulator.mostRecentPostCompactionUsage = null; + accumulator.mostRecentPostCompactionTimestamp = null; +} + +function accumulateTokenMetricEntry( + accumulator: TokenMetricAccumulator, + entry: TokenMetricEntry, + lastCompactBoundaryLineIndex: number +): void { + const { usage } = entry; + accumulator.inputTokens += usage.inputTokens; + accumulator.outputTokens += usage.outputTokens; + accumulator.cacheReadTokens += usage.cacheReadTokens; + accumulator.cacheCreationTokens += usage.cacheCreationTokens; + + if (!entry.isMainChain || !entry.timestamp) { + return; + } + + const entryTime = new Date(entry.timestamp); + if (!accumulator.mostRecentTimestamp || entryTime > accumulator.mostRecentTimestamp) { + accumulator.mostRecentTimestamp = entryTime; + accumulator.mostRecentMainChainUsage = usage; + } + if (entry.lineIndex > lastCompactBoundaryLineIndex + && (!accumulator.mostRecentPostCompactionTimestamp || entryTime > accumulator.mostRecentPostCompactionTimestamp)) { + accumulator.mostRecentPostCompactionTimestamp = entryTime; + accumulator.mostRecentPostCompactionUsage = usage; + } +} + function collectAgentIds(value: unknown, agentIds: Set) { if (!value || typeof value !== 'object') { return; @@ -149,12 +220,6 @@ export async function getTokenMetrics(transcriptPath: string): Promise { - const stopReason = entry.data.message?.stop_reason; - return Boolean(stopReason) || (stopReason === null && index === parsedEntries.length - 1); - }) - : parsedEntries; - - for (const { data, lineIndex: entryLineIndex } of entriesToCount) { - const usage = data.message?.usage; - if (!usage) { - continue; - } - - inputTokens += usage.input_tokens || 0; - outputTokens += usage.output_tokens || 0; - cacheReadTokens += usage.cache_read_input_tokens ?? 0; - cacheCreationTokens += usage.cache_creation_input_tokens ?? 0; - - // Track the most recent entry with isSidechain: false (or undefined, which defaults to main chain) - // Also skip API error messages (synthetic messages with 0 tokens) - if (data.isSidechain !== true && data.timestamp && !data.isApiErrorMessage) { - const entryTime = new Date(data.timestamp); - if (!mostRecentTimestamp || entryTime > mostRecentTimestamp) { - mostRecentTimestamp = entryTime; - mostRecentMainChainEntry = data; - } - if (entryLineIndex > lastCompactBoundaryLineIndex - && (!mostRecentPostCompactionTimestamp || entryTime > mostRecentPostCompactionTimestamp)) { - mostRecentPostCompactionTimestamp = entryTime; - mostRecentPostCompactionEntry = data; - } - } + if (hasStopReasonField && lastUsageEntry?.stopReason === null) { + accumulateTokenMetricEntry(streamingMetrics, lastUsageEntry, lastCompactBoundaryLineIndex); } + const metrics = hasStopReasonField ? streamingMetrics : legacyMetrics; + // Context length is the live occupancy of the current context window. // Without a compaction it is the most recent main-chain turn. After a // compaction, prefer the first turn following the boundary, then the // boundary's reported post-compaction size, and otherwise 0 - the stale // pre-compaction turn must never leak through. - const contextLengthFromEntry = (entry: TranscriptLine | null): number | null => { - const usage = entry?.message?.usage; + const contextLengthFromUsage = (usage: RetainedTokenUsage | null): number | null => { if (!usage) { return null; } - return (usage.input_tokens || 0) - + (usage.cache_read_input_tokens ?? 0) - + (usage.cache_creation_input_tokens ?? 0); + return usage.inputTokens + usage.cacheReadTokens + usage.cacheCreationTokens; }; - contextLength = lastCompactBoundaryLineIndex >= 0 - ? (contextLengthFromEntry(mostRecentPostCompactionEntry) ?? lastCompactBoundaryPostTokens ?? 0) - : (contextLengthFromEntry(mostRecentMainChainEntry) ?? 0); + const contextLength = lastCompactBoundaryLineIndex >= 0 + ? (contextLengthFromUsage(metrics.mostRecentPostCompactionUsage) ?? lastCompactBoundaryPostTokens ?? 0) + : (contextLengthFromUsage(metrics.mostRecentMainChainUsage) ?? 0); - const cachedTokens = cacheReadTokens + cacheCreationTokens; - const totalTokens = inputTokens + outputTokens + cachedTokens; + const cachedTokens = metrics.cacheReadTokens + metrics.cacheCreationTokens; + const totalTokens = metrics.inputTokens + metrics.outputTokens + cachedTokens; - return { inputTokens, outputTokens, cachedTokens, cacheReadTokens, cacheCreationTokens, totalTokens, contextLength }; + return { + inputTokens: metrics.inputTokens, + outputTokens: metrics.outputTokens, + cachedTokens, + cacheReadTokens: metrics.cacheReadTokens, + cacheCreationTokens: metrics.cacheCreationTokens, + totalTokens, + contextLength + }; } catch { return { inputTokens: 0, outputTokens: 0, cachedTokens: 0, cacheReadTokens: 0, cacheCreationTokens: 0, totalTokens: 0, contextLength: 0 }; } From 7732d8a65fce577e6eab8006e9b62a25217b6e09 Mon Sep 17 00:00:00 2001 From: Matthew Breedlove Date: Thu, 3 Sep 2026 00:30:51 -0400 Subject: [PATCH 3/4] fix transcript streaming edge cases --- src/ccstatusline.ts | 73 +- src/types/RenderContext.ts | 2 + src/utils/__tests__/jsonl-lines.test.ts | 87 ++- src/utils/__tests__/jsonl-metrics.test.ts | 131 +++- src/utils/compaction.ts | 80 ++- src/utils/jsonl-blocks.ts | 10 +- src/utils/jsonl-lines.ts | 226 +++++-- src/utils/jsonl-metadata.ts | 65 +- src/utils/jsonl-metrics.ts | 669 +++++++++++-------- src/utils/jsonl-session.ts | 34 + src/utils/jsonl.ts | 7 +- src/widgets/CacheTimer.ts | 3 +- src/widgets/SessionName.ts | 34 +- src/widgets/ThinkingEffort.ts | 6 +- src/widgets/__tests__/SessionName.test.ts | 70 +- src/widgets/__tests__/ThinkingEffort.test.ts | 17 +- 16 files changed, 990 insertions(+), 524 deletions(-) create mode 100644 src/utils/jsonl-session.ts diff --git a/src/ccstatusline.ts b/src/ccstatusline.ts index be492891..d5c756f1 100644 --- a/src/ccstatusline.ts +++ b/src/ccstatusline.ts @@ -2,20 +2,13 @@ import chalk from 'chalk'; import { runTUI } from './tui'; -import type { - SkillsMetrics, - SpeedMetrics, - TokenMetrics -} from './types'; +import type { SkillsMetrics } from './types'; import type { RenderContext } from './types/RenderContext'; import type { StatusJSON } from './types/StatusJSON'; import { StatusJSONSchema } from './types/StatusJSON'; import { getVisibleText } from './utils/ansi'; import { updateColorMap } from './utils/colors'; -import { - ZERO_COMPACTION_STATS, - getCompactionStats -} from './utils/compaction'; +import { ZERO_COMPACTION_STATS } from './utils/compaction'; import { getConfigLoadError, initConfigPath, @@ -27,11 +20,7 @@ import { refreshGitReviewCacheFromCli } from './utils/git-review-cache'; import { handleHookInput } from './utils/hook-handler'; -import { - getSessionDuration, - getSpeedMetricsCollection, - getTokenMetrics -} from './utils/jsonl'; +import { getTranscriptAnalysis } from './utils/jsonl'; import { advanceGlobalPowerlineThemeIndex } from './utils/powerline-theme-index'; import { buildConfigWarningBadge, @@ -117,6 +106,11 @@ async function renderMultipleLines(data: StatusJSON) { const speedWidgetTypes = new Set(['output-speed', 'input-speed', 'total-speed']); const hasSpeedItems = lines.some(line => line.some(item => speedWidgetTypes.has(item.type))); + const hasCompactionWidget = lines.some(line => line.some(item => item.type === 'compaction-counter')); + const hasThinkingEffortWidget = lines.some(line => line.some(item => item.type === 'thinking-effort')); + const hasSessionNameWidget = lines.some(line => line.some(item => item.type === 'session-name')); + const needsTranscriptThinkingEffort = hasThinkingEffortWidget + && (!data.effort || !('level' in data.effort)); const requestedSpeedWindows = new Set(); for (const line of lines) { for (const item of line) { @@ -126,39 +120,34 @@ async function renderMultipleLines(data: StatusJSON) { } } - let tokenMetrics: TokenMetrics | null = null; - if (data.transcript_path) { - tokenMetrics = await getTokenMetrics(data.transcript_path); - } - - let sessionDuration: string | null = null; - if (hasSessionClock && !hasSessionDurationInStatusJson(data) && data.transcript_path) { - sessionDuration = await getSessionDuration(data.transcript_path); - } - - const usageData = await prefetchUsageDataIfNeeded(lines, data); - - let speedMetrics: SpeedMetrics | null = null; - let windowedSpeedMetrics: Record | null = null; - if (hasSpeedItems && data.transcript_path) { - const speedMetricsCollection = await getSpeedMetricsCollection(data.transcript_path, { + const transcriptAnalysisPromise = data.transcript_path + ? getTranscriptAnalysis(data.transcript_path, { + includeSessionDuration: hasSessionClock && !hasSessionDurationInStatusJson(data), + includeSpeedMetrics: hasSpeedItems, includeSubagents: true, - windowSeconds: Array.from(requestedSpeedWindows) - }); - - speedMetrics = speedMetricsCollection.sessionAverage; - windowedSpeedMetrics = speedMetricsCollection.windowed; - } + speedWindowSeconds: Array.from(requestedSpeedWindows), + includeCompactionStats: hasCompactionWidget, + includeThinkingEffort: needsTranscriptThinkingEffort, + includeSessionName: hasSessionNameWidget + }) + : Promise.resolve(null); + const [transcriptAnalysis, usageData] = await Promise.all([ + transcriptAnalysisPromise, + prefetchUsageDataIfNeeded(lines, data) + ]); + + const tokenMetrics = transcriptAnalysis?.tokenMetrics ?? null; + const sessionDuration = transcriptAnalysis?.sessionDuration ?? null; + const speedMetrics = transcriptAnalysis?.speedMetricsCollection?.sessionAverage ?? null; + const windowedSpeedMetrics = transcriptAnalysis?.speedMetricsCollection?.windowed ?? null; let skillsMetrics: SkillsMetrics | null = null; if (data.session_id) { skillsMetrics = getSkillsMetrics(data.session_id); } - // Compaction stats — parse compact_boundary markers in this session's transcript - const hasCompactionWidget = lines.some(line => line.some(item => item.type === 'compaction-counter')); const compactionData = hasCompactionWidget - ? (data.transcript_path ? await getCompactionStats(data.transcript_path) : ZERO_COMPACTION_STATS) + ? (transcriptAnalysis?.compactionData ?? ZERO_COMPACTION_STATS) : null; // Create render context @@ -169,6 +158,12 @@ async function renderMultipleLines(data: StatusJSON) { windowedSpeedMetrics, usageData, sessionDuration, + transcriptSessionName: hasSessionNameWidget + ? (transcriptAnalysis?.sessionName ?? null) + : undefined, + transcriptThinkingEffort: needsTranscriptThinkingEffort + ? (transcriptAnalysis?.thinkingEffort ?? null) + : undefined, skillsMetrics, compactionData, terminalWidth: getTerminalWidth(), diff --git a/src/types/RenderContext.ts b/src/types/RenderContext.ts index 0a04489c..cbccca50 100644 --- a/src/types/RenderContext.ts +++ b/src/types/RenderContext.ts @@ -39,6 +39,8 @@ export interface RenderContext { windowedSpeedMetrics?: Record | null; usageData?: RenderUsageData | null; sessionDuration?: string | null; + transcriptSessionName?: string | null; + transcriptThinkingEffort?: { value: string; known: boolean } | null; blockMetrics?: BlockMetrics | null; skillsMetrics?: SkillsMetrics | null; compactionData?: CompactionData | null; diff --git a/src/utils/__tests__/jsonl-lines.test.ts b/src/utils/__tests__/jsonl-lines.test.ts index a380d152..b60bc613 100644 --- a/src/utils/__tests__/jsonl-lines.test.ts +++ b/src/utils/__tests__/jsonl-lines.test.ts @@ -10,9 +10,12 @@ import { } from 'vitest'; import { + JSONL_READ_CHUNK_BYTES, clearJsonlLineCache, iterateJsonlLines, + iterateJsonlLinesReverseSync, iterateJsonlLinesSync, + parseJsonlLine, readJsonlLines, readJsonlLinesSync } from '../jsonl-lines'; @@ -20,7 +23,12 @@ import { describe('jsonl line streaming', () => { const tempRoots: string[] = []; + beforeEach(() => { + clearJsonlLineCache(); + }); + afterEach(() => { + clearJsonlLineCache(); while (tempRoots.length > 0) { const root = tempRoots.pop(); if (root) { @@ -49,7 +57,7 @@ describe('jsonl line streaming', () => { '{"id":2}', '{"id":3}' ]); - expect(readJsonlLinesSync(filePath)).toEqual([ + expect(readJsonlLinesSync(filePath, { cache: false })).toEqual([ '{"id":1}', '{"id":2}', '{"id":3}' @@ -63,7 +71,7 @@ describe('jsonl line streaming', () => { '{"a":1}', '{"b":2}' ]); - expect(readJsonlLinesSync(filePath)).toEqual([ + expect(readJsonlLinesSync(filePath, { cache: false })).toEqual([ '{"a":1}', '{"b":2}' ]); @@ -74,10 +82,9 @@ describe('jsonl line streaming', () => { tempRoots.push(root); const filePath = path.join(root, 'utf8.jsonl'); - const chunkBytes = 1024 * 1024; const opening = '{"value":"'; const emoji = '😀'; - const line = `${opening}${'x'.repeat(chunkBytes - Buffer.byteLength(opening) - 2)}${emoji}"}`; + const line = `${opening}${'x'.repeat(JSONL_READ_CHUNK_BYTES - Buffer.byteLength(opening) - 2)}${emoji}"}`; fs.writeFileSync(filePath, line, 'utf8'); const lines = readJsonlLinesSync(filePath); @@ -86,17 +93,73 @@ describe('jsonl line streaming', () => { it('reads a record spanning many sync chunks followed by another record', () => { const filePath = writeTranscript('long-record.jsonl', [ - `{"value":"${'x'.repeat(6 * 1024 * 1024)}"}`, + `{"value":"${'x'.repeat(6 * JSONL_READ_CHUNK_BYTES)}"}`, '{"value":"next"}' ].join('\n')); const lines = Array.from(iterateJsonlLinesSync(filePath)); expect(lines).toHaveLength(2); - expect(nth(lines, 0)).toHaveLength((6 * 1024 * 1024) + 12); + expect(nth(lines, 0)).toHaveLength((6 * JSONL_READ_CHUNK_BYTES) + 12); expect(nth(lines, 1)).toBe('{"value":"next"}'); }); + it('uses LF-only record boundaries consistently in async and sync readers', async () => { + const unicodeRecord = JSON.stringify({ value: 'before\u2028middle\u2029after' }); + const filePath = writeTranscript('unicode-separators.jsonl', `${unicodeRecord}\n{"value":"next"}\n`); + + const asyncLines = await readJsonlLines(filePath, { cache: false }); + const syncLines = readJsonlLinesSync(filePath, { cache: false }); + + expect(asyncLines).toEqual([unicodeRecord, '{"value":"next"}']); + expect(syncLines).toEqual(asyncLines); + expect(asyncLines.map(parseJsonlLine)).not.toContain(null); + }); + + it('preserves lone carriage returns as content rather than record boundaries', async () => { + const filePath = writeTranscript('lone-cr.jsonl', 'left\rright\nnext'); + + await expect(readJsonlLines(filePath, { cache: false })).resolves.toEqual([ + 'left\rright', + 'next' + ]); + expect(readJsonlLinesSync(filePath, { cache: false })).toEqual(['left\rright', 'next']); + }); + + it('strips a UTF-8 BOM from the first record in both readers', async () => { + const filePath = writeTranscript('bom.jsonl', '\uFEFF{"value":1}\n{"value":2}\n'); + + const asyncLines = await readJsonlLines(filePath, { cache: false }); + const syncLines = readJsonlLinesSync(filePath, { cache: false }); + + expect(asyncLines).toEqual(['{"value":1}', '{"value":2}']); + expect(syncLines).toEqual(asyncLines); + expect(asyncLines.map(parseJsonlLine)).not.toContain(null); + }); + + it('reads records from newest to oldest without loading earlier content', () => { + const filePath = writeTranscript('reverse.jsonl', '\uFEFF{"value":1}\r\n{"value":2}\n{"value":3}'); + + expect(Array.from(iterateJsonlLinesReverseSync(filePath))).toEqual([ + '{"value":3}', + '{"value":2}', + '{"value":1}' + ]); + }); + + it('reverse-reads a UTF-8 record spanning multiple chunks', () => { + const opening = '{"value":"'; + const longLine = `${opening}${'x'.repeat((2 * JSONL_READ_CHUNK_BYTES) - Buffer.byteLength(opening) - 2)}😀"}`; + const filePath = writeTranscript('reverse-long.jsonl', `${longLine}\n{"value":"latest"}`); + + const lines = Array.from(iterateJsonlLinesReverseSync(filePath)); + + expect(lines).toEqual([ + '{"value":"latest"}', + longLine + ]); + }); + it('streams via async iterator without loading the full file as one string', async () => { const filePath = writeTranscript('stream.jsonl', [ '{"line":1}', @@ -117,11 +180,7 @@ describe('jsonl line streaming', () => { expect(Array.from(iterateJsonlLinesSync(filePath))).toEqual(seen); }); - it('can read files larger than Node max string length via streaming', async () => { - // Node refuses to create a single string longer than ~0x1fffffe8 (~512MB). - // Building a real 512MB+ fixture is too heavy for unit tests, so we prove - // the streaming path never calls readFile/readFileSync for the payload and - // still aggregates many chunks correctly by reading a multi-chunk file. + it('streams files containing many records across multiple chunks', async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-lines-')); tempRoots.push(root); const filePath = path.join(root, 'chunked.jsonl'); @@ -144,6 +203,12 @@ describe('jsonl line streaming', () => { const syncLines = readJsonlLinesSync(filePath); expect(syncLines).toHaveLength(lineCount); }, 30000); + + it('rejects stream open errors through the async reader', async () => { + const missingPath = path.join(os.tmpdir(), 'ccstatusline-jsonl-lines-missing', 'missing.jsonl'); + + await expect(readJsonlLines(missingPath, { cache: false })).rejects.toThrow(); + }); }); /** Matches MAX_CACHED_FILES in jsonl-lines.ts. */ diff --git a/src/utils/__tests__/jsonl-metrics.test.ts b/src/utils/__tests__/jsonl-metrics.test.ts index 344761ea..f7a367fa 100644 --- a/src/utils/__tests__/jsonl-metrics.test.ts +++ b/src/utils/__tests__/jsonl-metrics.test.ts @@ -12,7 +12,8 @@ import { getSessionDuration, getSpeedMetrics, getSpeedMetricsCollection, - getTokenMetrics + getTokenMetrics, + getTranscriptAnalysis } from '../jsonl'; function makeUsageLine(params: { @@ -163,6 +164,31 @@ describe('jsonl transcript metrics', () => { }); }); + it('ignores invalid timestamps when choosing the latest context usage', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); + tempRoots.push(root); + const transcriptPath = path.join(root, 'invalid-timestamp.jsonl'); + + fs.writeFileSync(transcriptPath, [ + makeUsageLine({ + timestamp: 'not-a-timestamp', + input: 100, + output: 1 + }), + makeUsageLine({ + timestamp: '2026-01-01T10:00:00.000Z', + input: 5, + output: 2, + cacheRead: 5000 + }) + ].join('\n')); + + const metrics = await getTokenMetrics(transcriptPath); + + expect(metrics.contextLength).toBe(5005); + expect(metrics.totalTokens).toBe(5108); + }); + it('skips intermediate streaming entries and only counts final entries per API call', async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); tempRoots.push(root); @@ -513,14 +539,13 @@ describe('jsonl transcript metrics', () => { }); }); - it('aggregates token metrics by streaming many usage lines without a whole-file string read', async () => { + it('aggregates token metrics across many usage records', async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); tempRoots.push(root); const transcriptPath = path.join(root, 'streamed-tokens.jsonl'); - // Many small lines so the streamer crosses chunk boundaries while the - // cumulative totals stay easy to assert. Regression for #550: full-file - // utf-8 reads throw once transcripts exceed Node's max string length. + // Many small lines keep the cumulative totals easy to assert while + // exercising repeated record aggregation. const lineCount = 2500; const handle = fs.openSync(transcriptPath, 'w'); try { @@ -551,7 +576,7 @@ describe('jsonl transcript metrics', () => { }); }, 30000); - it('discards large assistant content after extracting token fields', async () => { + it('aggregates usage rows containing multi-megabyte assistant content', async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); tempRoots.push(root); const transcriptPath = path.join(root, 'large-assistant-content.jsonl'); @@ -597,6 +622,100 @@ describe('jsonl transcript metrics', () => { }); }); + it('collects configured transcript metrics in one combined analysis', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); + tempRoots.push(root); + const transcriptPath = path.join(root, 'combined.jsonl'); + + fs.writeFileSync(transcriptPath, [ + JSON.stringify({ + type: 'user', + timestamp: '2026-01-01T10:00:00.000Z', + message: { content: 'hello' } + }), + JSON.stringify({ + type: 'assistant', + timestamp: '2026-01-01T10:01:00.000Z', + message: { + stop_reason: 'end_turn', + usage: { + input_tokens: 2, + output_tokens: 3, + cache_read_input_tokens: 4, + cache_creation_input_tokens: 1 + } + } + }), + JSON.stringify({ + type: 'custom-title', + customTitle: 'Combined Session', + timestamp: '2026-01-01T10:02:00.000Z' + }), + JSON.stringify({ + type: 'system', + timestamp: '2026-01-01T10:03:00.000Z', + message: { content: 'Set effort level to high' } + }), + JSON.stringify({ + type: 'system', + subtype: 'compact_boundary', + timestamp: '2026-01-01T10:04:00.000Z', + compactMetadata: { + trigger: 'manual', + preTokens: 10, + postTokens: 5 + } + }) + ].join('\n')); + + const analysis = await getTranscriptAnalysis(transcriptPath, { + includeSessionDuration: true, + includeSpeedMetrics: true, + speedWindowSeconds: [300], + includeCompactionStats: true, + includeThinkingEffort: true, + includeSessionName: true + }); + + expect(analysis).toEqual({ + tokenMetrics: { + inputTokens: 2, + outputTokens: 3, + cachedTokens: 5, + cacheReadTokens: 4, + cacheCreationTokens: 1, + totalTokens: 10, + contextLength: 5 + }, + sessionDuration: '4m', + speedMetricsCollection: { + sessionAverage: { + totalDurationMs: 60000, + inputTokens: 2, + outputTokens: 3, + totalTokens: 5, + requestCount: 1 + }, + windowed: { + 300: { + totalDurationMs: 60000, + inputTokens: 2, + outputTokens: 3, + totalTokens: 5, + requestCount: 1 + } + } + }, + compactionData: { + count: 1, + byTrigger: { auto: 0, manual: 1, unknown: 0 }, + tokensReclaimed: 5 + }, + thinkingEffort: { value: 'high', known: true }, + sessionName: 'Combined Session' + }); + }); + it('calculates speed metrics from user-to-assistant processing windows', async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-speed-')); tempRoots.push(root); diff --git a/src/utils/compaction.ts b/src/utils/compaction.ts index 7355f890..701dd96c 100644 --- a/src/utils/compaction.ts +++ b/src/utils/compaction.ts @@ -3,8 +3,8 @@ import * as fs from 'fs'; import type { CompactionData } from '../types/RenderContext'; import { - parseJsonlLine, - readJsonlLines + iterateJsonlLines, + parseJsonlLine } from './jsonl-lines'; /** Shared zeroed stats for missing/unreadable transcripts and as a render fallback. Treat as read-only. */ @@ -38,6 +38,42 @@ export function getCompactBoundaryPostTokens(record: unknown): number | null { return typeof post === 'number' && Number.isFinite(post) ? Math.max(0, post) : null; } +export function createCompactionStats(): CompactionData { + return { + count: 0, + byTrigger: { auto: 0, manual: 0, unknown: 0 }, + tokensReclaimed: 0 + }; +} + +export function accumulateCompactionStats(stats: CompactionData, record: unknown): void { + if (!isCompactBoundary(record)) { + return; + } + + stats.count += 1; + const meta = (record as { compactMetadata?: unknown }).compactMetadata; + const metaRecord = (typeof meta === 'object' && meta !== null) ? meta as Record : null; + + const trigger = metaRecord?.trigger; + if (trigger === 'auto') { + stats.byTrigger.auto += 1; + } else if (trigger === 'manual') { + stats.byTrigger.manual += 1; + } else { + stats.byTrigger.unknown += 1; + } + + const pre = metaRecord?.preTokens; + const post = metaRecord?.postTokens; + if (typeof pre === 'number' && typeof post === 'number') { + const reclaimed = pre - post; + if (Number.isFinite(reclaimed)) { + stats.tokensReclaimed += Math.max(0, reclaimed); + } + } +} + /** * Count context-compaction events and summarize their `compactMetadata` by * scanning the transcript for `{type:'system', subtype:'compact_boundary'}` @@ -49,38 +85,9 @@ export function getCompactBoundaryPostTokens(record: unknown): number | null { * are finite numbers; older markers without `postTokens` contribute 0. */ export function computeCompactionStats(lines: readonly string[]): CompactionData { - const stats: CompactionData = { - count: 0, - byTrigger: { auto: 0, manual: 0, unknown: 0 }, - tokensReclaimed: 0 - }; + const stats = createCompactionStats(); for (const line of lines) { - const record = parseJsonlLine(line); - if (!isCompactBoundary(record)) { - continue; - } - stats.count += 1; - - const meta = (record as { compactMetadata?: unknown }).compactMetadata; - const metaRecord = (typeof meta === 'object' && meta !== null) ? meta as Record : null; - - const trigger = metaRecord?.trigger; - if (trigger === 'auto') { - stats.byTrigger.auto += 1; - } else if (trigger === 'manual') { - stats.byTrigger.manual += 1; - } else { - stats.byTrigger.unknown += 1; - } - - const pre = metaRecord?.preTokens; - const post = metaRecord?.postTokens; - if (typeof pre === 'number' && typeof post === 'number') { - const reclaimed = pre - post; - if (Number.isFinite(reclaimed)) { - stats.tokensReclaimed += Math.max(0, reclaimed); - } - } + accumulateCompactionStats(stats, parseJsonlLine(line)); } return stats; } @@ -91,8 +98,11 @@ export async function getCompactionStats(transcriptPath: string): Promise(); @@ -69,6 +69,106 @@ function writeCached(identity: string, version: string, lines: readonly string[] return lines; } +/** + * Splits byte chunks using JSONL's LF delimiter without interpreting Unicode + * line separators or lone carriage returns as record boundaries. + */ +class JsonlLineSplitter { + private decoder = new StringDecoder('utf8'); + private readonly fragments: string[] = []; + private hasBytesInLine = false; + private isFirstLine = true; + + * write(chunk: Buffer): Generator { + let start = 0; + let newline = chunk.indexOf(0x0a, start); + + while (newline !== -1) { + this.append(chunk.subarray(start, newline)); + const line = this.finishLine(); + if (line !== null) { + yield line; + } + + start = newline + 1; + newline = chunk.indexOf(0x0a, start); + } + + this.append(chunk.subarray(start)); + } + + * end(): Generator { + if (!this.hasBytesInLine) { + return; + } + + const line = this.finishLine(); + if (line !== null) { + yield line; + } + } + + private append(bytes: Buffer): void { + if (bytes.length === 0) { + return; + } + + this.hasBytesInLine = true; + const decoded = this.decoder.write(bytes); + if (decoded.length > 0) { + this.fragments.push(decoded); + } + } + + private finishLine(): string | null { + const decodedTail = this.decoder.end(); + if (decodedTail.length > 0) { + this.fragments.push(decodedTail); + } + + let line = this.fragments.join(''); + this.fragments.length = 0; + this.decoder = new StringDecoder('utf8'); + this.hasBytesInLine = false; + + if (this.isFirstLine) { + this.isFirstLine = false; + if (line.charCodeAt(0) === 0xfeff) { + line = line.slice(1); + } + } + + if (line.endsWith('\r')) { + line = line.slice(0, -1); + } + + return line.length > 0 ? line : null; + } +} + +function decodeReverseLine(segments: Buffer[], totalBytes: number, stripBom: boolean): string | null { + let lineBuffer: Buffer; + if (segments.length === 1) { + const segment = segments[0]; + if (!segment) { + return null; + } + lineBuffer = segment; + } else { + lineBuffer = Buffer.concat(segments.slice().reverse(), totalBytes); + } + + let line = lineBuffer.toString('utf8'); + if (stripBom && line.charCodeAt(0) === 0xfeff) { + line = line.slice(1); + } + if (line.endsWith('\r')) { + line = line.slice(0, -1); + } + + return line.length > 0 ? line : null; +} + /** * Stream a JSONL file line-by-line without materializing the whole file as one * string. Claude Code session transcripts can exceed Node's max string length @@ -76,35 +176,37 @@ function writeCached(identity: string, version: string, lines: readonly string[] * catch the error end up reporting zeros. */ export async function* iterateJsonlLines(filePath: string): AsyncGenerator { - const stream = fs.createReadStream(filePath, { encoding: 'utf8' }); - const reader = createInterface({ - input: stream, - crlfDelay: Infinity - }); + const stream = fs.createReadStream(filePath, { highWaterMark: JSONL_READ_CHUNK_BYTES }); + const splitter = new JsonlLineSplitter(); + // Active read errors reject the async iterator. This listener also covers a + // late close error emitted after iterator cleanup on network filesystems. + stream.on('error', () => undefined); try { - for await (const line of reader) { - if (line.length > 0) { + for await (const chunk of stream as AsyncIterable) { + for (const line of splitter.write(chunk)) { yield line; } } + + for (const line of splitter.end()) { + yield line; + } } finally { - reader.close(); stream.destroy(); } } /** * Synchronous line iterator for call sites that cannot be async. - * Buffers chunk segments until a line is complete, then decodes the assembled - * Buffer so multi-byte UTF-8 sequences are never split across chunk boundaries. + * Decodes chunk segments incrementally so multi-byte UTF-8 sequences and long + * records spanning chunks remain correct without repeatedly copying prefixes. */ export function* iterateJsonlLinesSync(filePath: string): Generator { const fd = fs.openSync(filePath, 'r'); try { - const scratch = Buffer.allocUnsafe(SYNC_READ_CHUNK_BYTES); - const pending: Buffer[] = []; - let pendingBytes = 0; + const scratch = Buffer.allocUnsafe(JSONL_READ_CHUNK_BYTES); + const splitter = new JsonlLineSplitter(); for (;;) { const bytesRead = fs.readSync(fd, scratch, 0, scratch.length, null); @@ -112,52 +214,68 @@ export function* iterateJsonlLinesSync(filePath: string): Generator { break; } - const chunk = scratch.subarray(0, bytesRead); - let start = 0; + for (const line of splitter.write(scratch.subarray(0, bytesRead))) { + yield line; + } + } - for (let i = 0; i < chunk.length; i++) { - if (chunk[i] !== 0x0a) { - continue; - } + for (const line of splitter.end()) { + yield line; + } + } finally { + fs.closeSync(fd); + } +} - const segment = chunk.subarray(start, i); - let lineBuf: Buffer; - if (pending.length === 0) { - lineBuf = segment; - } else { - if (segment.length > 0) { - pending.push(segment); - pendingBytes += segment.length; - } - lineBuf = Buffer.concat(pending, pendingBytes); +/** + * Reads JSONL records from newest to oldest without loading the whole file. + * This is intended for widgets that only need the latest matching record. + */ +export function* iterateJsonlLinesReverseSync(filePath: string): Generator { + const fd = fs.openSync(filePath, 'r'); + try { + let position = fs.fstatSync(fd).size; + const segments: Buffer[] = []; + let totalBytes = 0; + + while (position > 0) { + const readSize = Math.min(JSONL_READ_CHUNK_BYTES, position); + position -= readSize; + + const chunk = Buffer.allocUnsafe(readSize); + fs.readSync(fd, chunk, 0, readSize, position); + let end = chunk.length; + let newline = chunk.lastIndexOf(0x0a, end - 1); + + while (newline !== -1) { + const segment = chunk.subarray(newline + 1, end); + if (segment.length > 0) { + segments.push(segment); + totalBytes += segment.length; } - pending.length = 0; - pendingBytes = 0; - start = i + 1; - - if (lineBuf.length > 0 && lineBuf[lineBuf.length - 1] === 0x0d) { - lineBuf = lineBuf.subarray(0, lineBuf.length - 1); - } - if (lineBuf.length > 0) { - yield lineBuf.toString('utf8'); + const line = decodeReverseLine(segments, totalBytes, false); + segments.length = 0; + totalBytes = 0; + if (line !== null) { + yield line; } + + end = newline; + newline = chunk.lastIndexOf(0x0a, end - 1); } - if (start < chunk.length) { - const remainder = Buffer.from(chunk.subarray(start)); - pending.push(remainder); - pendingBytes += remainder.length; + if (end > 0) { + const segment = chunk.subarray(0, end); + segments.push(segment); + totalBytes += segment.length; } } - if (pendingBytes > 0) { - let lineBuf = Buffer.concat(pending, pendingBytes); - if (lineBuf[lineBuf.length - 1] === 0x0d) { - lineBuf = lineBuf.subarray(0, lineBuf.length - 1); - } - if (lineBuf.length > 0) { - yield lineBuf.toString('utf8'); + if (totalBytes > 0) { + const line = decodeReverseLine(segments, totalBytes, true); + if (line !== null) { + yield line; } } } finally { diff --git a/src/utils/jsonl-metadata.ts b/src/utils/jsonl-metadata.ts index 0f6a22d2..48a1cc41 100644 --- a/src/utils/jsonl-metadata.ts +++ b/src/utils/jsonl-metadata.ts @@ -1,7 +1,7 @@ import { getVisibleText } from './ansi'; import { - parseJsonlLine, - readJsonlLinesSync + iterateJsonlLinesReverseSync, + parseJsonlLine } from './jsonl-lines'; const KNOWN_THINKING_EFFORTS = ['low', 'medium', 'high', 'xhigh', 'max'] as const; @@ -21,6 +21,8 @@ const UNKNOWN_EFFORT_PATTERN = /^(?=.*[a-z0-9])[a-z0-9-]{2,20}$/; interface TranscriptEntry { message?: { content?: string } } +export interface ThinkingEffortUpdate { effort: ResolvedThinkingEffort | undefined } + export function normalizeThinkingEffort(value: string | undefined): ResolvedThinkingEffort | undefined { if (!value) { return undefined; @@ -38,40 +40,47 @@ export function normalizeThinkingEffort(value: string | undefined): ResolvedThin return undefined; } -export function getTranscriptThinkingEffort(transcriptPath: string | undefined): ResolvedThinkingEffort | undefined { - if (!transcriptPath) { - return undefined; +/** + * Returns an update when a transcript record authoritatively changes the + * effort level. A /model result without an effort clears an older transcript + * value, matching the reverse-search behavior used by the widget fallback. + */ +export function getThinkingEffortUpdate(record: unknown): ThinkingEffortUpdate | null { + const entry = record as TranscriptEntry | null; + if (typeof entry?.message?.content !== 'string') { + return null; } - try { - const lines = readJsonlLinesSync(transcriptPath); + const content = entry.message.content; + if (!content.includes(EFFORT_STDOUT_PREFIX) && !content.includes(MODEL_STDOUT_PREFIX)) { + return null; + } - for (let i = lines.length - 1; i >= 0; i--) { - const line = lines[i]; - if (!line) { - continue; - } + const visibleContent = getVisibleText(content).trim(); + if (visibleContent.startsWith(EFFORT_STDOUT_PREFIX)) { + const effortMatch = EFFORT_STDOUT_REGEX.exec(visibleContent); + return effortMatch ? { effort: normalizeThinkingEffort(effortMatch[1]) } : null; + } - const entry = parseJsonlLine(line) as TranscriptEntry | null; - if (typeof entry?.message?.content !== 'string') { - continue; - } + if (!visibleContent.startsWith(MODEL_STDOUT_PREFIX)) { + return null; + } - const visibleContent = getVisibleText(entry.message.content).trim(); + const match = MODEL_STDOUT_EFFORT_REGEX.exec(visibleContent); + return { effort: normalizeThinkingEffort(match?.[1]) }; +} - if (visibleContent.startsWith(EFFORT_STDOUT_PREFIX)) { - const effortMatch = EFFORT_STDOUT_REGEX.exec(visibleContent); - if (effortMatch) { - return normalizeThinkingEffort(effortMatch[1]); - } - } +export function getTranscriptThinkingEffort(transcriptPath: string | undefined): ResolvedThinkingEffort | undefined { + if (!transcriptPath) { + return undefined; + } - if (!visibleContent.startsWith(MODEL_STDOUT_PREFIX)) { - continue; + try { + for (const line of iterateJsonlLinesReverseSync(transcriptPath)) { + const update = getThinkingEffortUpdate(parseJsonlLine(line)); + if (update) { + return update.effort; } - - const match = MODEL_STDOUT_EFFORT_REGEX.exec(visibleContent); - return normalizeThinkingEffort(match?.[1]); } } catch { return undefined; diff --git a/src/utils/jsonl-metrics.ts b/src/utils/jsonl-metrics.ts index 705914ee..6e30b63a 100644 --- a/src/utils/jsonl-metrics.ts +++ b/src/utils/jsonl-metrics.ts @@ -6,23 +6,30 @@ import type { TokenMetrics, TranscriptLine } from '../types'; +import type { CompactionData } from '../types/RenderContext'; import { + accumulateCompactionStats, + createCompactionStats, getCompactBoundaryPostTokens, isCompactBoundary } from './compaction'; import { iterateJsonlLines, - parseJsonlLine, - readJsonlLines + parseJsonlLine } from './jsonl-lines'; +import { + getThinkingEffortUpdate, + type ResolvedThinkingEffort +} from './jsonl-metadata'; +import { getSessionNameFromRecord } from './jsonl-session'; export interface SpeedMetricsOptions { includeSubagents?: boolean; windowSeconds?: number; } -interface SpeedMetricsCollectionOptions { +export interface SpeedMetricsCollectionOptions { includeSubagents?: boolean; windowSeconds?: number[]; } @@ -32,6 +39,36 @@ export interface SpeedMetricsCollection { windowed: Record; } +export interface TranscriptAnalysisOptions { + includeSessionDuration?: boolean; + includeSpeedMetrics?: boolean; + includeSubagents?: boolean; + speedWindowSeconds?: number[]; + includeCompactionStats?: boolean; + includeThinkingEffort?: boolean; + includeSessionName?: boolean; +} + +export interface TranscriptAnalysis { + tokenMetrics: TokenMetrics; + sessionDuration: string | null; + speedMetricsCollection: SpeedMetricsCollection | null; + compactionData: CompactionData | null; + thinkingEffort: ResolvedThinkingEffort | undefined; + sessionName: string | null; +} + +interface TranscriptScanOptions extends TranscriptAnalysisOptions { includeTokenMetrics?: boolean } + +interface TranscriptScanResult { + tokenMetrics: TokenMetrics | null; + sessionDuration: string | null; + speedMetricsCollection: SpeedMetricsCollection | null; + compactionData: CompactionData | null; + thinkingEffort: ResolvedThinkingEffort | undefined; + sessionName: string | null; +} + interface SpeedInterval { startMs: number; endMs: number; @@ -59,9 +96,8 @@ interface RetainedTokenUsage { interface TokenMetricEntry { usage: RetainedTokenUsage; stopReason: string | null | undefined; - timestamp: string | undefined; + timestampMs: number | null; isMainChain: boolean; - lineIndex: number; } interface TokenMetricAccumulator { @@ -70,9 +106,30 @@ interface TokenMetricAccumulator { cacheReadTokens: number; cacheCreationTokens: number; mostRecentMainChainUsage: RetainedTokenUsage | null; - mostRecentTimestamp: Date | null; + mostRecentTimestampMs: number | null; mostRecentPostCompactionUsage: RetainedTokenUsage | null; - mostRecentPostCompactionTimestamp: Date | null; + mostRecentPostCompactionTimestampMs: number | null; +} + +interface TokenMetricState { + metrics: TokenMetricAccumulator; + hasStopReasonField: boolean; + lastUsageEntry: TokenMetricEntry | null; + sawCompactBoundary: boolean; + boundaryAfterLastUsage: boolean; + lastCompactBoundaryPostTokens: number | null; +} + +function createEmptyTokenMetrics(): TokenMetrics { + return { + inputTokens: 0, + outputTokens: 0, + cachedTokens: 0, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalTokens: 0, + contextLength: 0 + }; } function createTokenMetricAccumulator(): TokenMetricAccumulator { @@ -82,21 +139,21 @@ function createTokenMetricAccumulator(): TokenMetricAccumulator { cacheReadTokens: 0, cacheCreationTokens: 0, mostRecentMainChainUsage: null, - mostRecentTimestamp: null, + mostRecentTimestampMs: null, mostRecentPostCompactionUsage: null, - mostRecentPostCompactionTimestamp: null + mostRecentPostCompactionTimestampMs: null }; } function resetPostCompactionUsage(accumulator: TokenMetricAccumulator): void { accumulator.mostRecentPostCompactionUsage = null; - accumulator.mostRecentPostCompactionTimestamp = null; + accumulator.mostRecentPostCompactionTimestampMs = null; } function accumulateTokenMetricEntry( accumulator: TokenMetricAccumulator, entry: TokenMetricEntry, - lastCompactBoundaryLineIndex: number + includePostCompactionUsage: boolean ): void { const { usage } = entry; accumulator.inputTokens += usage.inputTokens; @@ -104,22 +161,94 @@ function accumulateTokenMetricEntry( accumulator.cacheReadTokens += usage.cacheReadTokens; accumulator.cacheCreationTokens += usage.cacheCreationTokens; - if (!entry.isMainChain || !entry.timestamp) { + if (!entry.isMainChain || entry.timestampMs === null) { return; } - const entryTime = new Date(entry.timestamp); - if (!accumulator.mostRecentTimestamp || entryTime > accumulator.mostRecentTimestamp) { - accumulator.mostRecentTimestamp = entryTime; + if (accumulator.mostRecentTimestampMs === null || entry.timestampMs > accumulator.mostRecentTimestampMs) { + accumulator.mostRecentTimestampMs = entry.timestampMs; accumulator.mostRecentMainChainUsage = usage; } - if (entry.lineIndex > lastCompactBoundaryLineIndex - && (!accumulator.mostRecentPostCompactionTimestamp || entryTime > accumulator.mostRecentPostCompactionTimestamp)) { - accumulator.mostRecentPostCompactionTimestamp = entryTime; + if (includePostCompactionUsage + && (accumulator.mostRecentPostCompactionTimestampMs === null + || entry.timestampMs > accumulator.mostRecentPostCompactionTimestampMs)) { + accumulator.mostRecentPostCompactionTimestampMs = entry.timestampMs; accumulator.mostRecentPostCompactionUsage = usage; } } +function createTokenMetricState(): TokenMetricState { + return { + metrics: createTokenMetricAccumulator(), + hasStopReasonField: false, + lastUsageEntry: null, + sawCompactBoundary: false, + boundaryAfterLastUsage: false, + lastCompactBoundaryPostTokens: null + }; +} + +function collectTokenMetricRecord(state: TokenMetricState, data: TranscriptLine | null, timestampMs: number | null): void { + const compactBoundary = isCompactBoundary(data); + if (compactBoundary) { + state.sawCompactBoundary = true; + state.boundaryAfterLastUsage = true; + state.lastCompactBoundaryPostTokens = getCompactBoundaryPostTokens(data); + resetPostCompactionUsage(state.metrics); + } + + const message = data?.message; + const usage = message?.usage; + if (usage) { + const entry: TokenMetricEntry = { + usage: { + inputTokens: usage.input_tokens || 0, + outputTokens: usage.output_tokens || 0, + cacheReadTokens: usage.cache_read_input_tokens ?? 0, + cacheCreationTokens: usage.cache_creation_input_tokens ?? 0 + }, + stopReason: message.stop_reason, + timestampMs, + isMainChain: data?.isSidechain !== true && !data?.isApiErrorMessage + }; + + const hasStopReason = Object.hasOwn(message, 'stop_reason'); + if (hasStopReason && !state.hasStopReasonField) { + state.hasStopReasonField = true; + state.metrics = createTokenMetricAccumulator(); + } + if (!state.hasStopReasonField || entry.stopReason) { + accumulateTokenMetricEntry(state.metrics, entry, !compactBoundary); + } + state.lastUsageEntry = entry; + state.boundaryAfterLastUsage = compactBoundary; + } +} + +function finishTokenMetrics(state: TokenMetricState): TokenMetrics { + if (state.hasStopReasonField && state.lastUsageEntry?.stopReason === null) { + accumulateTokenMetricEntry(state.metrics, state.lastUsageEntry, !state.boundaryAfterLastUsage); + } + + const contextLengthFromUsage = (usage: RetainedTokenUsage | null): number | null => usage + ? usage.inputTokens + usage.cacheReadTokens + usage.cacheCreationTokens + : null; + const contextLength = state.sawCompactBoundary + ? (contextLengthFromUsage(state.metrics.mostRecentPostCompactionUsage) ?? state.lastCompactBoundaryPostTokens ?? 0) + : (contextLengthFromUsage(state.metrics.mostRecentMainChainUsage) ?? 0); + const cachedTokens = state.metrics.cacheReadTokens + state.metrics.cacheCreationTokens; + + return { + inputTokens: state.metrics.inputTokens, + outputTokens: state.metrics.outputTokens, + cachedTokens, + cacheReadTokens: state.metrics.cacheReadTokens, + cacheCreationTokens: state.metrics.cacheCreationTokens, + totalTokens: state.metrics.inputTokens + state.metrics.outputTokens + cachedTokens, + contextLength + }; +} + function collectAgentIds(value: unknown, agentIds: Set) { if (!value || typeof value !== 'object') { return; @@ -142,191 +271,23 @@ function collectAgentIds(value: unknown, agentIds: Set) { } } -function getReferencedSubagentIds(lines: readonly string[]): Set { - const agentIds = new Set(); - - for (const line of lines) { - const data = parseJsonlLine(line); - if (!data) { - continue; - } - - collectAgentIds(data, agentIds); - } - - return agentIds; -} - export async function getSessionDuration(transcriptPath: string): Promise { - try { - if (!fs.existsSync(transcriptPath)) { - return null; - } - - let firstTimestamp: Date | null = null; - let lastTimestamp: Date | null = null; - let sawAnyLine = false; - - for await (const line of iterateJsonlLines(transcriptPath)) { - sawAnyLine = true; - const data = parseJsonlLine(line) as { timestamp?: string } | null; - if (!data?.timestamp) { - continue; - } - - const timestamp = new Date(data.timestamp); - if (Number.isNaN(timestamp.getTime())) { - continue; - } - - firstTimestamp ??= timestamp; - lastTimestamp = timestamp; - } - - if (!sawAnyLine || !firstTimestamp || !lastTimestamp) { - return null; - } - - // Calculate duration in milliseconds - const durationMs = lastTimestamp.getTime() - firstTimestamp.getTime(); - - // Convert to minutes - const totalMinutes = Math.floor(durationMs / (1000 * 60)); - - if (totalMinutes < 1) { - return '<1m'; - } - - const hours = Math.floor(totalMinutes / 60); - const minutes = totalMinutes % 60; - - if (hours === 0) { - return `${minutes}m`; - } else if (minutes === 0) { - return `${hours}hr`; - } else { - return `${hours}hr ${minutes}m`; - } - } catch { - return null; - } + const result = await scanTranscript(transcriptPath, { includeSessionDuration: true }); + return result.sessionDuration; } export async function getTokenMetrics(transcriptPath: string): Promise { - try { - // Stream line-by-line. Full-file readFile('utf-8') throws once a session - // transcript exceeds Node's max string length (~512MB). - if (!fs.existsSync(transcriptPath)) { - return { inputTokens: 0, outputTokens: 0, cachedTokens: 0, cacheReadTokens: 0, cacheCreationTokens: 0, totalTokens: 0, contextLength: 0 }; - } - - // Parse each line and sum up token usage for totals. - // Claude Code writes multiple JSONL entries per API call during streaming: - // intermediate entries have stop_reason: null, and the final entry has a - // string value like "end_turn" or "tool_use". For streaming-aware - // transcripts, count finalized entries plus the latest unfinished entry so - // live updates do not overcount duplicate partial rows. If the transcript - // format has no stop_reason field at all, fall back to counting all entries. - // - // Claude Code also writes a { type:'system', subtype:'compact_boundary' } - // record on every compaction. Usage entries before the most recent boundary - // describe a context that no longer exists, so they must not drive context - // length - otherwise it stays stuck at the pre-compaction size until the - // next turn repopulates Claude Code's live status data. - let lastCompactBoundaryLineIndex = -1; - let lastCompactBoundaryPostTokens: number | null = null; - - // Aggregate both transcript formats in one pass. This avoids retaining - // assistant message content while preserving the legacy fallback when no - // usage row has a stop_reason field. - const legacyMetrics = createTokenMetricAccumulator(); - const streamingMetrics = createTokenMetricAccumulator(); - let lastUsageEntry: TokenMetricEntry | null = null; - let hasStopReasonField = false; - let lineIndex = 0; - - for await (const line of iterateJsonlLines(transcriptPath)) { - const data = parseJsonlLine(line) as TranscriptLine | null; - if (isCompactBoundary(data)) { - lastCompactBoundaryLineIndex = lineIndex; - lastCompactBoundaryPostTokens = getCompactBoundaryPostTokens(data); - resetPostCompactionUsage(legacyMetrics); - resetPostCompactionUsage(streamingMetrics); - } - const message = data?.message; - const usage = message?.usage; - if (usage) { - const entry: TokenMetricEntry = { - usage: { - inputTokens: usage.input_tokens || 0, - outputTokens: usage.output_tokens || 0, - cacheReadTokens: usage.cache_read_input_tokens ?? 0, - cacheCreationTokens: usage.cache_creation_input_tokens ?? 0 - }, - stopReason: message.stop_reason, - timestamp: data?.timestamp, - isMainChain: data?.isSidechain !== true && !data?.isApiErrorMessage, - lineIndex - }; - - accumulateTokenMetricEntry(legacyMetrics, entry, lastCompactBoundaryLineIndex); - if (Object.hasOwn(message, 'stop_reason')) { - hasStopReasonField = true; - } - if (entry.stopReason) { - accumulateTokenMetricEntry(streamingMetrics, entry, lastCompactBoundaryLineIndex); - } - lastUsageEntry = entry; - } - lineIndex += 1; - } - - if (hasStopReasonField && lastUsageEntry?.stopReason === null) { - accumulateTokenMetricEntry(streamingMetrics, lastUsageEntry, lastCompactBoundaryLineIndex); - } - - const metrics = hasStopReasonField ? streamingMetrics : legacyMetrics; - - // Context length is the live occupancy of the current context window. - // Without a compaction it is the most recent main-chain turn. After a - // compaction, prefer the first turn following the boundary, then the - // boundary's reported post-compaction size, and otherwise 0 - the stale - // pre-compaction turn must never leak through. - const contextLengthFromUsage = (usage: RetainedTokenUsage | null): number | null => { - if (!usage) { - return null; - } - return usage.inputTokens + usage.cacheReadTokens + usage.cacheCreationTokens; - }; - - const contextLength = lastCompactBoundaryLineIndex >= 0 - ? (contextLengthFromUsage(metrics.mostRecentPostCompactionUsage) ?? lastCompactBoundaryPostTokens ?? 0) - : (contextLengthFromUsage(metrics.mostRecentMainChainUsage) ?? 0); - - const cachedTokens = metrics.cacheReadTokens + metrics.cacheCreationTokens; - const totalTokens = metrics.inputTokens + metrics.outputTokens + cachedTokens; - - return { - inputTokens: metrics.inputTokens, - outputTokens: metrics.outputTokens, - cachedTokens, - cacheReadTokens: metrics.cacheReadTokens, - cacheCreationTokens: metrics.cacheCreationTokens, - totalTokens, - contextLength - }; - } catch { - return { inputTokens: 0, outputTokens: 0, cachedTokens: 0, cacheReadTokens: 0, cacheCreationTokens: 0, totalTokens: 0, contextLength: 0 }; - } + const result = await scanTranscript(transcriptPath, { includeTokenMetrics: true }); + return result.tokenMetrics ?? createEmptyTokenMetrics(); } -function parseTimestamp(value: string | undefined): Date | null { +function parseTimestampMs(value: string | undefined): number | null { if (!value) { return null; } - const timestamp = new Date(value); - return Number.isNaN(timestamp.getTime()) ? null : timestamp; + const timestampMs = Date.parse(value); + return Number.isNaN(timestampMs) ? null : timestampMs; } function mergeIntervals(intervals: SpeedInterval[]): SpeedInterval[] { @@ -383,60 +344,60 @@ function normalizeWindowSeconds(value: number | undefined): number | null { return normalized > 0 ? normalized : null; } -function collectSpeedMetricsFromLines(lines: readonly string[], ignoreSidechain: boolean): CollectedSpeedMetrics { - const requests: SpeedRequest[] = []; +interface SpeedMetricCollectorState extends CollectedSpeedMetrics { lastUserTimestampMs: number | null } - let lastUserTimestamp: Date | null = null; - let latestTimestampMs: number | null = null; +function createSpeedMetricCollector(): SpeedMetricCollectorState { + return { + requests: [], + latestTimestampMs: null, + lastUserTimestampMs: null + }; +} - for (const line of lines) { - const data = parseJsonlLine(line) as TranscriptLine | null; - if (!data || data.isApiErrorMessage) { - continue; - } +function collectSpeedMetricRecord( + state: SpeedMetricCollectorState, + data: TranscriptLine | null, + timestampMs: number | null, + ignoreSidechain: boolean +): void { + if (!data || data.isApiErrorMessage || (ignoreSidechain && data.isSidechain === true)) { + return; + } - if (ignoreSidechain && data.isSidechain === true) { - continue; - } + if (timestampMs !== null && (state.latestTimestampMs === null || timestampMs > state.latestTimestampMs)) { + state.latestTimestampMs = timestampMs; + } - const entryTimestamp = parseTimestamp(data.timestamp); - if (entryTimestamp) { - const entryTimestampMs = entryTimestamp.getTime(); - if (latestTimestampMs === null || entryTimestampMs > latestTimestampMs) { - latestTimestampMs = entryTimestampMs; - } - } + if (data.type === 'user' && timestampMs !== null) { + state.lastUserTimestampMs = timestampMs; + return; + } - if (data.type === 'user' && entryTimestamp) { - lastUserTimestamp = entryTimestamp; - continue; - } + if (data.type !== 'assistant' || !data.message?.usage) { + return; + } - if (data.type === 'assistant' && data.message?.usage) { - const inputTokens = data.message.usage.input_tokens || 0; - const outputTokens = data.message.usage.output_tokens || 0; - let interval: SpeedInterval | null = null; - if (entryTimestamp && lastUserTimestamp) { - const startMs = lastUserTimestamp.getTime(); - const endMs = entryTimestamp.getTime(); - if (endMs > startMs) { - interval = { startMs, endMs }; - } - } + let interval: SpeedInterval | null = null; + if (timestampMs !== null && state.lastUserTimestampMs !== null && timestampMs > state.lastUserTimestampMs) { + interval = { startMs: state.lastUserTimestampMs, endMs: timestampMs }; + } - requests.push({ - inputTokens, - outputTokens, - assistantTimestampMs: entryTimestamp ? entryTimestamp.getTime() : null, - interval - }); - } + state.requests.push({ + inputTokens: data.message.usage.input_tokens || 0, + outputTokens: data.message.usage.output_tokens || 0, + assistantTimestampMs: timestampMs, + interval + }); +} + +async function collectSpeedMetricsFromFile(filePath: string, ignoreSidechain: boolean): Promise { + const state = createSpeedMetricCollector(); + for await (const line of iterateJsonlLines(filePath)) { + const data = parseJsonlLine(line) as TranscriptLine | null; + collectSpeedMetricRecord(state, data, parseTimestampMs(data?.timestamp), ignoreSidechain); } - return { - requests, - latestTimestampMs - }; + return state; } function mergeCollectedSpeedMetrics(parts: CollectedSpeedMetrics[]): CollectedSpeedMetrics { @@ -527,6 +488,155 @@ function buildEmptyWindowedMetrics(windowSeconds: number[]): Record = {}; + for (const window of windowSeconds) { + windowed[window.toString()] = buildSpeedMetrics(combined, window); + } + + return { + sessionAverage: buildSpeedMetrics(combined), + windowed + }; +} + +function formatSessionDuration(firstTimestampMs: number | null, lastTimestampMs: number | null): string | null { + if (firstTimestampMs === null || lastTimestampMs === null) { + return null; + } + + const totalMinutes = Math.floor((lastTimestampMs - firstTimestampMs) / (1000 * 60)); + if (totalMinutes < 1) { + return '<1m'; + } + + const hours = Math.floor(totalMinutes / 60); + const minutes = totalMinutes % 60; + if (hours === 0) { + return `${minutes}m`; + } + if (minutes === 0) { + return `${hours}hr`; + } + return `${hours}hr ${minutes}m`; +} + +function normalizeSpeedWindows(windowSeconds: number[] | undefined): number[] { + return Array.from( + new Set( + (windowSeconds ?? []) + .map(window => normalizeWindowSeconds(window)) + .filter((window): window is number => window !== null) + ) + ); +} + +function createEmptyScanResult(options: TranscriptScanOptions, speedWindows: number[]): TranscriptScanResult { + return { + tokenMetrics: options.includeTokenMetrics ? createEmptyTokenMetrics() : null, + sessionDuration: null, + speedMetricsCollection: options.includeSpeedMetrics + ? { + sessionAverage: createEmptySpeedMetrics(), + windowed: buildEmptyWindowedMetrics(speedWindows) + } + : null, + compactionData: options.includeCompactionStats ? createCompactionStats() : null, + thinkingEffort: undefined, + sessionName: null + }; +} + +async function scanTranscript(transcriptPath: string, options: TranscriptScanOptions): Promise { + const speedWindows = normalizeSpeedWindows(options.speedWindowSeconds); + const emptyResult = createEmptyScanResult(options, speedWindows); + if (!fs.existsSync(transcriptPath)) { + return emptyResult; + } + + const tokenState = options.includeTokenMetrics ? createTokenMetricState() : null; + const speedState = options.includeSpeedMetrics ? createSpeedMetricCollector() : null; + const compactionData = options.includeCompactionStats ? createCompactionStats() : null; + const referencedAgentIds = options.includeSpeedMetrics && options.includeSubagents + ? new Set() + : null; + let firstTimestampMs: number | null = null; + let lastTimestampMs: number | null = null; + let thinkingEffort: ResolvedThinkingEffort | undefined; + let sessionName: string | null = null; + + try { + for await (const line of iterateJsonlLines(transcriptPath)) { + const data = parseJsonlLine(line) as TranscriptLine | null; + const needsTimestamp = options.includeSessionDuration === true + || speedState !== null + || Boolean(data?.message?.usage); + const timestampMs = needsTimestamp ? parseTimestampMs(data?.timestamp) : null; + + if (tokenState) { + collectTokenMetricRecord(tokenState, data, timestampMs); + } + if (options.includeSessionDuration && timestampMs !== null) { + firstTimestampMs ??= timestampMs; + lastTimestampMs = timestampMs; + } + if (speedState) { + collectSpeedMetricRecord(speedState, data, timestampMs, true); + } + if (referencedAgentIds) { + collectAgentIds(data, referencedAgentIds); + } + if (compactionData) { + accumulateCompactionStats(compactionData, data); + } + if (options.includeThinkingEffort) { + const update = getThinkingEffortUpdate(data); + if (update) { + thinkingEffort = update.effort; + } + } + if (options.includeSessionName) { + sessionName = getSessionNameFromRecord(data) ?? sessionName; + } + } + + let speedMetricsCollection: SpeedMetricsCollection | null = null; + if (speedState) { + const collected: CollectedSpeedMetrics[] = [speedState]; + if (referencedAgentIds) { + const subagentPaths = getSubagentTranscriptPaths(transcriptPath, referencedAgentIds); + const subagentMetrics = await Promise.all(subagentPaths.map(async (subagentPath) => { + try { + return await collectSpeedMetricsFromFile(subagentPath, false); + } catch { + return null; + } + })); + for (const metrics of subagentMetrics) { + if (metrics) { + collected.push(metrics); + } + } + } + speedMetricsCollection = buildSpeedMetricsCollection(collected, speedWindows); + } + + return { + tokenMetrics: tokenState ? finishTokenMetrics(tokenState) : null, + sessionDuration: options.includeSessionDuration + ? formatSessionDuration(firstTimestampMs, lastTimestampMs) + : null, + speedMetricsCollection, + compactionData, + thinkingEffort, + sessionName + }; + } catch { + return emptyResult; + } +} + function getSubagentTranscriptPaths(transcriptPath: string, referencedAgentIds: Set): string[] { if (referencedAgentIds.size === 0) { return []; @@ -582,65 +692,34 @@ export async function getSpeedMetricsCollection( transcriptPath: string, options: SpeedMetricsCollectionOptions = {} ): Promise { - const normalizedWindows = Array.from( - new Set( - (options.windowSeconds ?? []) - .map(window => normalizeWindowSeconds(window)) - .filter((window): window is number => window !== null) - ) - ); - const emptyWindowedMetrics = buildEmptyWindowedMetrics(normalizedWindows); - - try { - if (!fs.existsSync(transcriptPath)) { - return { - sessionAverage: createEmptySpeedMetrics(), - windowed: emptyWindowedMetrics - }; - } - - const mainLines = await readJsonlLines(transcriptPath); - const allCollected: CollectedSpeedMetrics[] = [ - collectSpeedMetricsFromLines(mainLines, true) - ]; - - if (options.includeSubagents === true) { - const referencedSubagentIds = getReferencedSubagentIds(mainLines); - const subagentPaths = getSubagentTranscriptPaths(transcriptPath, referencedSubagentIds); - const subagentMetricsResults = await Promise.all(subagentPaths.map(async (subagentPath) => { - try { - const subagentLines = await readJsonlLines(subagentPath); - return collectSpeedMetricsFromLines(subagentLines, false); - } catch { - return null; - } - })); - - for (const subagentMetrics of subagentMetricsResults) { - if (!subagentMetrics) { - continue; - } - - allCollected.push(subagentMetrics); - } - } + const result = await scanTranscript(transcriptPath, { + includeSpeedMetrics: true, + includeSubagents: options.includeSubagents, + speedWindowSeconds: options.windowSeconds + }); + return result.speedMetricsCollection ?? { + sessionAverage: createEmptySpeedMetrics(), + windowed: buildEmptyWindowedMetrics(normalizeSpeedWindows(options.windowSeconds)) + }; +} - const combined = mergeCollectedSpeedMetrics(allCollected); - const windowed: Record = {}; - for (const window of normalizedWindows) { - windowed[window.toString()] = buildSpeedMetrics(combined, window); - } +export async function getTranscriptAnalysis( + transcriptPath: string, + options: TranscriptAnalysisOptions = {} +): Promise { + const result = await scanTranscript(transcriptPath, { + ...options, + includeTokenMetrics: true + }); - return { - sessionAverage: buildSpeedMetrics(combined), - windowed - }; - } catch { - return { - sessionAverage: createEmptySpeedMetrics(), - windowed: emptyWindowedMetrics - }; - } + return { + tokenMetrics: result.tokenMetrics ?? createEmptyTokenMetrics(), + sessionDuration: result.sessionDuration, + speedMetricsCollection: result.speedMetricsCollection, + compactionData: result.compactionData, + thinkingEffort: result.thinkingEffort, + sessionName: result.sessionName + }; } export async function getSpeedMetrics( diff --git a/src/utils/jsonl-session.ts b/src/utils/jsonl-session.ts new file mode 100644 index 00000000..01ddbb3d --- /dev/null +++ b/src/utils/jsonl-session.ts @@ -0,0 +1,34 @@ +import { + iterateJsonlLinesReverseSync, + parseJsonlLine +} from './jsonl-lines'; + +export function getSessionNameFromRecord(record: unknown): string | null { + if (typeof record !== 'object' || record === null) { + return null; + } + + const entry = record as { type?: unknown; customTitle?: unknown }; + return entry.type === 'custom-title' && typeof entry.customTitle === 'string' && entry.customTitle.length > 0 + ? entry.customTitle + : null; +} + +export function getTranscriptSessionName(transcriptPath: string | undefined): string | null { + if (!transcriptPath) { + return null; + } + + try { + for (const line of iterateJsonlLinesReverseSync(transcriptPath)) { + const sessionName = getSessionNameFromRecord(parseJsonlLine(line)); + if (sessionName !== null) { + return sessionName; + } + } + } catch { + return null; + } + + return null; +} diff --git a/src/utils/jsonl.ts b/src/utils/jsonl.ts index 3c6497a7..1b071eb0 100644 --- a/src/utils/jsonl.ts +++ b/src/utils/jsonl.ts @@ -9,7 +9,12 @@ export { getSessionDuration, getSpeedMetrics, getSpeedMetricsCollection, - getTokenMetrics + getTokenMetrics, + getTranscriptAnalysis +} from './jsonl-metrics'; +export type { + TranscriptAnalysis, + TranscriptAnalysisOptions } from './jsonl-metrics'; export { getTranscriptThinkingEffort, diff --git a/src/widgets/CacheTimer.ts b/src/widgets/CacheTimer.ts index 22bf3ec0..3d785bf0 100644 --- a/src/widgets/CacheTimer.ts +++ b/src/widgets/CacheTimer.ts @@ -75,8 +75,7 @@ function hasCacheActivity(entry: TranscriptEntry): boolean { // A single transcript record can exceed the initial tail read (pasted prompts // and tool results reach hundreds of KiB), leaving only an unparseable // fragment in view, so the read doubles until the state resolves or the whole -// file has been scanned — the same worst case as the full-file transcript -// reads the token widgets already do every render. +// file has been scanned. const INITIAL_TAIL_BYTES = 32768; /** diff --git a/src/widgets/SessionName.ts b/src/widgets/SessionName.ts index cc747c8b..f3bfb7e6 100644 --- a/src/widgets/SessionName.ts +++ b/src/widgets/SessionName.ts @@ -1,5 +1,3 @@ -import * as fs from 'fs'; - import type { RenderContext } from '../types/RenderContext'; import type { Settings } from '../types/Settings'; import type { @@ -7,6 +5,7 @@ import type { WidgetEditorDisplay, WidgetItem } from '../types/Widget'; +import { getTranscriptSessionName } from '../utils/jsonl-session'; export class SessionNameWidget implements Widget { getDefaultColor(): string { return 'cyan'; } @@ -22,35 +21,14 @@ export class SessionNameWidget implements Widget { return item.rawValue ? 'my-session' : 'Session: my-session'; } - const transcriptPath = context.data?.transcript_path; - if (!transcriptPath) { + const sessionName = context.transcriptSessionName === undefined + ? getTranscriptSessionName(context.data?.transcript_path) + : context.transcriptSessionName; + if (sessionName === null) { return null; } - try { - const content = fs.readFileSync(transcriptPath, 'utf-8'); - const lines = content.split('\n'); - - // Find the most recent custom-title entry (search from end) - for (let i = lines.length - 1; i >= 0; i--) { - const line = lines[i]?.trim(); - if (!line) - continue; - - try { - const entry = JSON.parse(line) as { type?: string; customTitle?: string }; - if (entry.type === 'custom-title' && entry.customTitle) { - return item.rawValue ? entry.customTitle : `Session: ${entry.customTitle}`; - } - } catch { - // Skip malformed lines - } - } - } catch { - // File not readable - } - - return null; + return item.rawValue ? sessionName : `Session: ${sessionName}`; } supportsRawValue(): boolean { return true; } diff --git a/src/widgets/ThinkingEffort.ts b/src/widgets/ThinkingEffort.ts index 53c54cbd..0d5ea4ea 100644 --- a/src/widgets/ThinkingEffort.ts +++ b/src/widgets/ThinkingEffort.ts @@ -41,7 +41,11 @@ function resolveThinkingEffort(context: RenderContext): ResolvedThinkingEffort | return statusEffort; } - return getTranscriptThinkingEffort(context.data?.transcript_path) + const transcriptEffort = context.transcriptThinkingEffort === undefined + ? getTranscriptThinkingEffort(context.data?.transcript_path) + : context.transcriptThinkingEffort ?? undefined; + + return transcriptEffort ?? resolveThinkingEffortFromSettings() ?? null; } diff --git a/src/widgets/__tests__/SessionName.test.ts b/src/widgets/__tests__/SessionName.test.ts index 07fe0765..f8b74c36 100644 --- a/src/widgets/__tests__/SessionName.test.ts +++ b/src/widgets/__tests__/SessionName.test.ts @@ -1,11 +1,12 @@ import * as fs from 'fs'; +import os from 'os'; +import path from 'path'; import { afterEach, beforeEach, describe, expect, - it, - vi + it } from 'vitest'; import type { @@ -15,12 +16,16 @@ import type { import { DEFAULT_SETTINGS } from '../../types/Settings'; import { SessionNameWidget } from '../SessionName'; -let mockReadFileSync: { mockImplementation: (fn: () => string | never) => void }; +let tempDir: string; function render(transcriptPath: string | undefined, fileContent: string | null, rawValue = false, isPreview = false) { const widget = new SessionNameWidget(); + const resolvedTranscriptPath = transcriptPath ? path.join(tempDir, 'session.jsonl') : undefined; + if (resolvedTranscriptPath && fileContent !== null) { + fs.writeFileSync(resolvedTranscriptPath, fileContent); + } const context: RenderContext = { - data: transcriptPath ? { transcript_path: transcriptPath } : undefined, + data: resolvedTranscriptPath ? { transcript_path: resolvedTranscriptPath } : undefined, isPreview }; const item: WidgetItem = { @@ -29,25 +34,16 @@ function render(transcriptPath: string | undefined, fileContent: string | null, rawValue }; - if (fileContent !== null) { - mockReadFileSync.mockImplementation(() => fileContent); - } else { - mockReadFileSync.mockImplementation(() => { - throw new Error('File not found'); - }); - } - return widget.render(item, context, DEFAULT_SETTINGS); } describe('SessionNameWidget', () => { beforeEach(() => { - vi.restoreAllMocks(); - mockReadFileSync = vi.spyOn(fs, 'readFileSync'); + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-session-name-')); }); afterEach(() => { - vi.restoreAllMocks(); + fs.rmSync(tempDir, { recursive: true, force: true }); }); it('should have session category', () => { @@ -99,9 +95,53 @@ describe('SessionNameWidget', () => { expect(result).toBe('Session: New Name'); }); + it('uses a session name precomputed by the shared transcript analysis', () => { + const widget = new SessionNameWidget(); + const result = widget.render( + { id: 'session-name', type: 'session-name' }, + { + data: { transcript_path: path.join(tempDir, 'missing.jsonl') }, + transcriptSessionName: 'Precomputed Session' + }, + DEFAULT_SETTINGS + ); + + expect(result).toBe('Session: Precomputed Session'); + }); + it('should skip malformed JSON lines', () => { const content = 'not valid json\n{"type":"custom-title","customTitle":"Valid Title"}'; const result = render('/some/path/session.jsonl', content); expect(result).toBe('Session: Valid Title'); }); + + it('reads the latest title from a transcript larger than Node maximum string length', () => { + const transcriptPath = path.join(tempDir, 'huge-session.jsonl'); + const handle = fs.openSync(transcriptPath, 'w'); + try { + fs.writeSync( + handle, + '\n{"type":"custom-title","customTitle":"Huge Session"}', + undefined, + 'utf8' + ); + fs.writeSync( + handle, + '\n{"type":"custom-title","customTitle":"Latest Huge Session"}', + 0x1fffffe8 + 1024, + 'utf8' + ); + } finally { + fs.closeSync(handle); + } + + const widget = new SessionNameWidget(); + const result = widget.render( + { id: 'session-name', type: 'session-name' }, + { data: { transcript_path: transcriptPath } }, + DEFAULT_SETTINGS + ); + + expect(result).toBe('Session: Latest Huge Session'); + }); }); diff --git a/src/widgets/__tests__/ThinkingEffort.test.ts b/src/widgets/__tests__/ThinkingEffort.test.ts index 9f8843b6..9b8c7d7f 100644 --- a/src/widgets/__tests__/ThinkingEffort.test.ts +++ b/src/widgets/__tests__/ThinkingEffort.test.ts @@ -56,6 +56,7 @@ function render(options: { isPreview?: boolean; statusData?: Partial; settingsValue?: unknown; + transcriptThinkingEffort?: RenderContext['transcriptThinkingEffort']; } = {}): string | null { const { transcriptPath = options.fileContent !== undefined ? path.join(tempDir, 'session.jsonl') : undefined, @@ -63,7 +64,8 @@ function render(options: { rawValue = false, isPreview = false, statusData = {}, - settingsValue = {} + settingsValue = {}, + transcriptThinkingEffort } = options; const widget = new ThinkingEffortWidget(); @@ -73,7 +75,8 @@ function render(options: { }; const context: RenderContext = { data: Object.keys(data).length > 0 ? data : undefined, - isPreview + isPreview, + transcriptThinkingEffort }; const item: WidgetItem = { id: 'thinking-effort', @@ -231,6 +234,16 @@ describe('ThinkingEffortWidget', () => { }); expect(result).toBe('Thinking: medium'); }); + + it('uses effort precomputed by the shared transcript analysis', () => { + const result = render({ + transcriptPath: path.join(tempDir, 'missing.jsonl'), + transcriptThinkingEffort: { value: 'high', known: true }, + settingsValue: { effortLevel: 'low' } + }); + + expect(result).toBe('Thinking: high'); + }); }); describe('/effort command source', () => { From dbe146bf15f98a109019dd61d4d0b1f4a448d2bf Mon Sep 17 00:00:00 2001 From: Matthew Breedlove Date: Thu, 3 Sep 2026 00:52:48 -0400 Subject: [PATCH 4/4] refactor(jsonl): share usage normalization and drop the unused line cache The transcript scan and the live status JSON each coerced usage counts their own way, so a malformed count could make the two disagree about the same session. context-window.ts now exports parseUsageTokens and contextLengthFromUsageTokens, and the token accumulator, the speed collector and the live path all go through them. Routing every consumer through the single scanTranscript pass left the materialize-and-cache layer with no production callers, so drop readJsonlLines/readJsonlLinesSync and the cache behind them, plus getCompactionStats (which carried a second full transcript walk), computeCompactionStats, getTokenMetrics, getSessionDuration, getSpeedMetrics and getSpeedMetricsCollection. getTranscriptAnalysis is now the only entry point. Their tests keep the same cases but exercise the live path instead of the removed wrappers. Also give the streaming test a real assertion: it spies on fs.readFile and fs.readFileSync rather than only claiming the file is never read as one string. Claude-Session: https://claude.ai/code/session_0153W4v1gsL5jo5sAg7VGmoQ --- src/utils/__tests__/compaction.test.ts | 30 ++- src/utils/__tests__/jsonl-lines.test.ts | 298 +++------------------- src/utils/__tests__/jsonl-metrics.test.ts | 84 +++++- src/utils/compaction.ts | 41 --- src/utils/context-window.ts | 25 +- src/utils/jsonl-lines.ts | 134 +--------- src/utils/jsonl-metrics.ts | 94 ++----- src/utils/jsonl.ts | 8 +- 8 files changed, 182 insertions(+), 532 deletions(-) diff --git a/src/utils/__tests__/compaction.test.ts b/src/utils/__tests__/compaction.test.ts index 9724da8f..9db62eff 100644 --- a/src/utils/__tests__/compaction.test.ts +++ b/src/utils/__tests__/compaction.test.ts @@ -9,11 +9,29 @@ import { it } from 'vitest'; +import type { CompactionData } from '../../types/RenderContext'; import { ZERO_COMPACTION_STATS, - computeCompactionStats, - getCompactionStats + accumulateCompactionStats, + createCompactionStats } from '../compaction'; +import { parseJsonlLine } from '../jsonl-lines'; +import { getTranscriptAnalysis } from '../jsonl-metrics'; + +/** Folds raw records through the same accumulator the transcript scan uses. */ +function computeCompactionStats(lines: readonly string[]): CompactionData { + const stats = createCompactionStats(); + for (const line of lines) { + accumulateCompactionStats(stats, parseJsonlLine(line)); + } + + return stats; +} + +async function compactionStatsFor(transcriptPath: string): Promise { + const analysis = await getTranscriptAnalysis(transcriptPath, { includeCompactionStats: true }); + return analysis.compactionData; +} describe('computeCompactionStats', () => { it('returns zeroed stats for no compaction markers', () => { @@ -122,7 +140,7 @@ describe('computeCompactionStats', () => { }); }); -describe('getCompactionStats', () => { +describe('compaction stats over a transcript', () => { let dir: string; beforeEach(() => { @@ -134,7 +152,7 @@ describe('getCompactionStats', () => { }); it('returns zeroed stats when the transcript file does not exist', async () => { - await expect(getCompactionStats(path.join(dir, 'missing.jsonl'))).resolves.toEqual(ZERO_COMPACTION_STATS); + await expect(compactionStatsFor(path.join(dir, 'missing.jsonl'))).resolves.toEqual(ZERO_COMPACTION_STATS); }); it('computes stats from a real-shaped transcript', async () => { @@ -146,7 +164,7 @@ describe('getCompactionStats', () => { JSON.stringify({ type: 'system', subtype: 'compact_boundary', content: 'Conversation compacted', compactMetadata: { trigger: 'auto', preTokens: 912661, postTokens: 30026 }, version: '2.1.161' }) ].join('\n') + '\n'; fs.writeFileSync(file, content); - await expect(getCompactionStats(file)).resolves.toEqual({ + await expect(compactionStatsFor(file)).resolves.toEqual({ count: 2, byTrigger: { auto: 1, manual: 1, unknown: 0 }, tokensReclaimed: (837327 - 25443) + (912661 - 30026) @@ -154,6 +172,6 @@ describe('getCompactionStats', () => { }); it('returns zeroed stats when the transcript path is not a readable file', async () => { - await expect(getCompactionStats(dir)).resolves.toEqual(ZERO_COMPACTION_STATS); + await expect(compactionStatsFor(dir)).resolves.toEqual(ZERO_COMPACTION_STATS); }); }); diff --git a/src/utils/__tests__/jsonl-lines.test.ts b/src/utils/__tests__/jsonl-lines.test.ts index b60bc613..ed6afb67 100644 --- a/src/utils/__tests__/jsonl-lines.test.ts +++ b/src/utils/__tests__/jsonl-lines.test.ts @@ -3,32 +3,38 @@ import os from 'os'; import path from 'path'; import { afterEach, - beforeEach, describe, expect, - it + it, + vi } from 'vitest'; import { JSONL_READ_CHUNK_BYTES, - clearJsonlLineCache, iterateJsonlLines, iterateJsonlLinesReverseSync, iterateJsonlLinesSync, - parseJsonlLine, - readJsonlLines, - readJsonlLinesSync + parseJsonlLine } from '../jsonl-lines'; +async function collectAsync(filePath: string): Promise { + const lines: string[] = []; + for await (const line of iterateJsonlLines(filePath)) { + lines.push(line); + } + + return lines; +} + +function collectSync(filePath: string): string[] { + return Array.from(iterateJsonlLinesSync(filePath)); +} + describe('jsonl line streaming', () => { const tempRoots: string[] = []; - beforeEach(() => { - clearJsonlLineCache(); - }); - afterEach(() => { - clearJsonlLineCache(); + vi.restoreAllMocks(); while (tempRoots.length > 0) { const root = tempRoots.pop(); if (root) { @@ -52,12 +58,12 @@ describe('jsonl line streaming', () => { '{"id":3}' ].join('\n')); - await expect(readJsonlLines(filePath)).resolves.toEqual([ + await expect(collectAsync(filePath)).resolves.toEqual([ '{"id":1}', '{"id":2}', '{"id":3}' ]); - expect(readJsonlLinesSync(filePath, { cache: false })).toEqual([ + expect(collectSync(filePath)).toEqual([ '{"id":1}', '{"id":2}', '{"id":3}' @@ -67,11 +73,11 @@ describe('jsonl line streaming', () => { it('skips empty lines like the previous whole-file trim/split path', async () => { const filePath = writeTranscript('empty-lines.jsonl', '\n{"a":1}\n\n{"b":2}\n\n'); - await expect(readJsonlLines(filePath)).resolves.toEqual([ + await expect(collectAsync(filePath)).resolves.toEqual([ '{"a":1}', '{"b":2}' ]); - expect(readJsonlLinesSync(filePath, { cache: false })).toEqual([ + expect(collectSync(filePath)).toEqual([ '{"a":1}', '{"b":2}' ]); @@ -87,7 +93,7 @@ describe('jsonl line streaming', () => { const line = `${opening}${'x'.repeat(JSONL_READ_CHUNK_BYTES - Buffer.byteLength(opening) - 2)}${emoji}"}`; fs.writeFileSync(filePath, line, 'utf8'); - const lines = readJsonlLinesSync(filePath); + const lines = collectSync(filePath); expect(lines).toEqual([line]); }); @@ -108,8 +114,8 @@ describe('jsonl line streaming', () => { const unicodeRecord = JSON.stringify({ value: 'before\u2028middle\u2029after' }); const filePath = writeTranscript('unicode-separators.jsonl', `${unicodeRecord}\n{"value":"next"}\n`); - const asyncLines = await readJsonlLines(filePath, { cache: false }); - const syncLines = readJsonlLinesSync(filePath, { cache: false }); + const asyncLines = await collectAsync(filePath); + const syncLines = collectSync(filePath); expect(asyncLines).toEqual([unicodeRecord, '{"value":"next"}']); expect(syncLines).toEqual(asyncLines); @@ -119,18 +125,18 @@ describe('jsonl line streaming', () => { it('preserves lone carriage returns as content rather than record boundaries', async () => { const filePath = writeTranscript('lone-cr.jsonl', 'left\rright\nnext'); - await expect(readJsonlLines(filePath, { cache: false })).resolves.toEqual([ + await expect(collectAsync(filePath)).resolves.toEqual([ 'left\rright', 'next' ]); - expect(readJsonlLinesSync(filePath, { cache: false })).toEqual(['left\rright', 'next']); + expect(collectSync(filePath)).toEqual(['left\rright', 'next']); }); it('strips a UTF-8 BOM from the first record in both readers', async () => { const filePath = writeTranscript('bom.jsonl', '\uFEFF{"value":1}\n{"value":2}\n'); - const asyncLines = await readJsonlLines(filePath, { cache: false }); - const syncLines = readJsonlLinesSync(filePath, { cache: false }); + const asyncLines = await collectAsync(filePath); + const syncLines = collectSync(filePath); expect(asyncLines).toEqual(['{"value":1}', '{"value":2}']); expect(syncLines).toEqual(asyncLines); @@ -160,24 +166,27 @@ describe('jsonl line streaming', () => { ]); }); - it('streams via async iterator without loading the full file as one string', async () => { + it('never reads the file as one string, which would throw past the max string length', async () => { const filePath = writeTranscript('stream.jsonl', [ '{"line":1}', '{"line":2}', '{"line":3}' ].join('\n')); - const seen: string[] = []; - for await (const line of iterateJsonlLines(filePath)) { - seen.push(line); - } - expect(seen).toEqual([ + const readFile = vi.spyOn(fs, 'readFile'); + const readFileSync = vi.spyOn(fs, 'readFileSync'); + + const asyncLines = await collectAsync(filePath); + const syncLines = collectSync(filePath); + + expect(asyncLines).toEqual([ '{"line":1}', '{"line":2}', '{"line":3}' ]); - - expect(Array.from(iterateJsonlLinesSync(filePath))).toEqual(seen); + expect(syncLines).toEqual(asyncLines); + expect(readFile).not.toHaveBeenCalled(); + expect(readFileSync).not.toHaveBeenCalled(); }); it('streams files containing many records across multiple chunks', async () => { @@ -195,29 +204,22 @@ describe('jsonl line streaming', () => { fs.closeSync(handle); } - const lines = await readJsonlLines(filePath); + const lines = await collectAsync(filePath); expect(lines).toHaveLength(lineCount); expect(nth(lines, 0)).toBe(`{"i":0,"pad":"${'z'.repeat(200)}"}`); expect(nth(lines, lineCount - 1)).toBe(`{"i":${lineCount - 1},"pad":"${'z'.repeat(200)}"}`); - const syncLines = readJsonlLinesSync(filePath); + const syncLines = collectSync(filePath); expect(syncLines).toHaveLength(lineCount); }, 30000); it('rejects stream open errors through the async reader', async () => { const missingPath = path.join(os.tmpdir(), 'ccstatusline-jsonl-lines-missing', 'missing.jsonl'); - await expect(readJsonlLines(missingPath, { cache: false })).rejects.toThrow(); + await expect(collectAsync(missingPath)).rejects.toThrow(); }); }); -/** Matches MAX_CACHED_FILES in jsonl-lines.ts. */ -const MAX_CACHED_FILES = 8; - -function makeLine(value: string): string { - return JSON.stringify({ value }); -} - /** Indexed access that fails loudly, since the config forbids non-null assertions. */ function nth(items: readonly T[], index: number): T { const item = items[index]; @@ -227,219 +229,3 @@ function nth(items: readonly T[], index: number): T { return item; } - -describe('jsonl line cache', () => { - let tempDir: string; - let transcript: string; - - function transcriptAt(name: string, contents: string): string { - const filePath = path.join(tempDir, name); - fs.writeFileSync(filePath, contents); - return filePath; - } - - /** Rewrites a file at a byte length and modification time of the caller's choosing. */ - function rewrite(filePath: string, contents: string, mtime: Date): void { - fs.writeFileSync(filePath, contents); - fs.utimesSync(filePath, mtime, mtime); - } - - beforeEach(() => { - tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-lines-')); - transcript = path.join(tempDir, 'session.jsonl'); - clearJsonlLineCache(); - }); - - afterEach(() => { - clearJsonlLineCache(); - fs.rmSync(tempDir, { recursive: true, force: true }); - }); - - it('splits a transcript into its non-empty lines', () => { - fs.writeFileSync(transcript, `${makeLine('a')}\n\n${makeLine('b')}\n`); - - expect(readJsonlLinesSync(transcript)).toEqual([makeLine('a'), makeLine('b')]); - }); - - it('reuses the split lines while the file is unchanged', () => { - fs.writeFileSync(transcript, `${makeLine('aa')}\n`); - - expect(readJsonlLinesSync(transcript)).toBe(readJsonlLinesSync(transcript)); - }); - - it('re-reads once the file grows', () => { - fs.writeFileSync(transcript, `${makeLine('a')}\n`); - readJsonlLinesSync(transcript); - - fs.appendFileSync(transcript, `${makeLine('b')}\n`); - - expect(readJsonlLinesSync(transcript)).toEqual([makeLine('a'), makeLine('b')]); - }); - - it('re-reads when only the size changes, at a pinned modification time', () => { - const pinned = new Date('2020-01-01T00:00:00Z'); - rewrite(transcript, `${makeLine('a')}\n`, pinned); - readJsonlLinesSync(transcript); - - // Same mtime, different length: only size can tell these apart. - rewrite(transcript, `${makeLine('a')}\n${makeLine('b')}\n`, pinned); - - expect(readJsonlLinesSync(transcript)).toEqual([makeLine('a'), makeLine('b')]); - }); - - it('re-reads when only the modification time changes, at a pinned size', () => { - rewrite(transcript, `${makeLine('aa')}\n`, new Date('2020-01-01T00:00:00Z')); - readJsonlLinesSync(transcript); - - // Same length, later mtime: only the timestamp can tell these apart. - rewrite(transcript, `${makeLine('bb')}\n`, new Date('2021-01-01T00:00:00Z')); - - expect(readJsonlLinesSync(transcript)).toEqual([makeLine('bb')]); - }); - - it('serves one entry for the many spellings of a path', () => { - fs.writeFileSync(transcript, `${makeLine('a')}\n`); - - // Built by hand rather than through path.join, which would normalize - // these back into the very string they are meant to differ from. - const base = path.basename(transcript); - const sep = path.sep; - const spellings = [ - transcript, - `${tempDir}${sep}.${sep}${base}`, - `${tempDir}${sep}${sep}${base}` - ]; - - // A backslash separates paths on Windows and is an ordinary filename - // character elsewhere, so only Windows has the slash-direction spelling. - if (process.platform === 'win32') { - spellings.push(transcript.split('\\').join('/')); - } - - const first = readJsonlLinesSync(nth(spellings, 0)); - - for (const spelling of spellings.slice(1)) { - expect(readJsonlLinesSync(spelling)).toBe(first); - } - }); - - it('shares one entry between the sync and async readers', async () => { - fs.writeFileSync(transcript, `${makeLine('aa')}\n`); - - const fromSync = readJsonlLinesSync(transcript); - - await expect(readJsonlLines(transcript)).resolves.toBe(fromSync); - }); - - it('re-reads after the cache is cleared', async () => { - fs.writeFileSync(transcript, `${makeLine('aa')}\n`); - const first = readJsonlLinesSync(transcript); - - clearJsonlLineCache(); - - const second = await readJsonlLines(transcript); - expect(second).not.toBe(first); - expect(second).toEqual(first); - }); - - it('caches each transcript separately', () => { - const other = transcriptAt('other.jsonl', `${makeLine('z')}\n`); - fs.writeFileSync(transcript, `${makeLine('a')}\n`); - - const first = readJsonlLinesSync(transcript); - readJsonlLinesSync(other); - - expect(readJsonlLinesSync(transcript)).toBe(first); - expect(readJsonlLinesSync(other)).toEqual([makeLine('z')]); - }); - - it('keeps a second transcript when a first is written', () => { - const other = transcriptAt('other.jsonl', `${makeLine('z')}\n`); - const kept = readJsonlLinesSync(other); - - fs.writeFileSync(transcript, `${makeLine('a')}\n`); - readJsonlLinesSync(transcript); - - expect(readJsonlLinesSync(other)).toBe(kept); - }); - - it('evicts the oldest entry once the cap is passed', () => { - const paths = Array.from({ length: MAX_CACHED_FILES + 1 }, (_unused, index) => transcriptAt(`t${index}.jsonl`, `${makeLine(`v${index}`)}\n`)); - const cached = paths.slice(0, MAX_CACHED_FILES).map(filePath => readJsonlLinesSync(filePath)); - - // One past the cap, which evicts the least recently written entry. - readJsonlLinesSync(nth(paths, MAX_CACHED_FILES)); - - // Assert the survivor before re-reading the evicted one, since that read re-inserts. - expect(readJsonlLinesSync(nth(paths, 1))).toBe(nth(cached, 1)); - expect(readJsonlLinesSync(nth(paths, 0))).not.toBe(nth(cached, 0)); - }); - - it('counts a rewritten entry as the most recently used', () => { - const paths = Array.from({ length: MAX_CACHED_FILES + 1 }, (_unused, index) => transcriptAt(`r${index}.jsonl`, `${makeLine(`v${index}`)}\n`)); - for (const filePath of paths.slice(0, MAX_CACHED_FILES)) { - readJsonlLinesSync(filePath); - } - - // Rewriting the oldest entry should move it off the eviction block. - fs.appendFileSync(nth(paths, 0), `${makeLine('grown')}\n`); - const refreshed = readJsonlLinesSync(nth(paths, 0)); - - readJsonlLinesSync(nth(paths, MAX_CACHED_FILES)); - - expect(readJsonlLinesSync(nth(paths, 0))).toBe(refreshed); - }); - - it('retains every entry up to the cap', () => { - const paths = Array.from({ length: MAX_CACHED_FILES }, (_unused, index) => transcriptAt(`k${index}.jsonl`, `${makeLine(`v${index}`)}\n`)); - const first = paths.map(filePath => readJsonlLinesSync(filePath)); - - paths.forEach((filePath, index) => { - expect(readJsonlLinesSync(filePath)).toBe(nth(first, index)); - }); - }); - - it('does not cache when caching is disabled', () => { - fs.writeFileSync(transcript, `${makeLine('a')}\n`); - - const first = readJsonlLinesSync(transcript, { cache: false }); - const second = readJsonlLinesSync(transcript, { cache: false }); - - expect(second).not.toBe(first); - expect(second).toEqual(first); - }); - - it('does not cache when caching is disabled on the async reader', async () => { - fs.writeFileSync(transcript, `${makeLine('a')}\n`); - - const first = await readJsonlLines(transcript, { cache: false }); - const second = await readJsonlLines(transcript, { cache: false }); - - expect(second).not.toBe(first); - expect(second).toEqual(first); - }); - - it('does not populate the cache from an uncached read', () => { - fs.writeFileSync(transcript, `${makeLine('a')}\n`); - - readJsonlLinesSync(transcript, { cache: false }); - - // A cached read after an uncached one still has to do its own work. - const cached = readJsonlLinesSync(transcript); - expect(readJsonlLinesSync(transcript)).toBe(cached); - }); - - it('propagates the read failure for a missing transcript', () => { - expect(() => readJsonlLinesSync(path.join(tempDir, 'absent.jsonl'))).toThrow(); - }); - - it('serves a recreated transcript rather than the deleted one', () => { - fs.writeFileSync(transcript, `${makeLine('a')}\n`); - readJsonlLinesSync(transcript); - - fs.rmSync(transcript); - fs.writeFileSync(transcript, `${makeLine('b')}\n${makeLine('c')}\n`); - - expect(readJsonlLinesSync(transcript)).toEqual([makeLine('b'), makeLine('c')]); - }); -}); diff --git a/src/utils/__tests__/jsonl-metrics.test.ts b/src/utils/__tests__/jsonl-metrics.test.ts index f7a367fa..348ea432 100644 --- a/src/utils/__tests__/jsonl-metrics.test.ts +++ b/src/utils/__tests__/jsonl-metrics.test.ts @@ -8,13 +8,57 @@ import { it } from 'vitest'; -import { - getSessionDuration, - getSpeedMetrics, - getSpeedMetricsCollection, - getTokenMetrics, - getTranscriptAnalysis -} from '../jsonl'; +import type { + SpeedMetrics, + TokenMetrics +} from '../../types'; +import type { StatusJSON } from '../../types/StatusJSON'; +import { getContextWindowMetrics } from '../context-window'; +import { getTranscriptAnalysis } from '../jsonl'; +import type { SpeedMetricsCollection } from '../jsonl-metrics'; + +// The render path makes one combined scan; these narrow the result to the one +// metric each case is about, so the assertions stay readable. +async function getSessionDuration(transcriptPath: string): Promise { + const analysis = await getTranscriptAnalysis(transcriptPath, { includeSessionDuration: true }); + return analysis.sessionDuration; +} + +async function getTokenMetrics(transcriptPath: string): Promise { + const analysis = await getTranscriptAnalysis(transcriptPath); + return analysis.tokenMetrics; +} + +async function getSpeedMetricsCollection( + transcriptPath: string, + options: { includeSubagents?: boolean; windowSeconds?: number[] } = {} +): Promise { + const analysis = await getTranscriptAnalysis(transcriptPath, { + includeSpeedMetrics: true, + includeSubagents: options.includeSubagents, + speedWindowSeconds: options.windowSeconds + }); + if (!analysis.speedMetricsCollection) { + throw new Error('speed metrics were requested but not collected'); + } + + return analysis.speedMetricsCollection; +} + +async function getSpeedMetrics( + transcriptPath: string, + options: { includeSubagents?: boolean; windowSeconds?: number } = {} +): Promise { + const { windowSeconds } = options; + const collection = await getSpeedMetricsCollection(transcriptPath, { + includeSubagents: options.includeSubagents, + windowSeconds: windowSeconds === undefined ? [] : [windowSeconds] + }); + + return windowSeconds === undefined + ? collection.sessionAverage + : collection.windowed[windowSeconds.toString()] ?? collection.sessionAverage; +} function makeUsageLine(params: { timestamp: string; @@ -78,6 +122,32 @@ describe('jsonl transcript metrics', () => { } }); + it('clamps malformed usage counts the way the live status path does', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); + tempRoots.push(root); + const transcriptPath = path.join(root, 'malformed-usage.jsonl'); + const usage = { + input_tokens: '12', + output_tokens: -7, + cache_read_input_tokens: 5000, + cache_creation_input_tokens: 100 + }; + fs.writeFileSync(transcriptPath, `${JSON.stringify({ + timestamp: '2026-01-01T10:00:00.000Z', + message: { stop_reason: 'end_turn', usage } + })}\n`); + + const metrics = await getTokenMetrics(transcriptPath); + const live = getContextWindowMetrics({ context_window: { current_usage: usage } } as unknown as StatusJSON); + + // A non-numeric or negative count reads as 0 in both paths, so the + // transcript fallback can never disagree with the live status JSON. + expect(metrics.inputTokens).toBe(0); + expect(metrics.outputTokens).toBe(0); + expect(metrics.contextLength).toBe(5100); + expect(metrics.contextLength).toBe(live.contextLengthTokens); + }); + it('formats session duration as <1m for sub-minute transcripts', async () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ccstatusline-jsonl-metrics-')); tempRoots.push(root); diff --git a/src/utils/compaction.ts b/src/utils/compaction.ts index 701dd96c..fbbd4632 100644 --- a/src/utils/compaction.ts +++ b/src/utils/compaction.ts @@ -1,12 +1,5 @@ -import * as fs from 'fs'; - import type { CompactionData } from '../types/RenderContext'; -import { - iterateJsonlLines, - parseJsonlLine -} from './jsonl-lines'; - /** Shared zeroed stats for missing/unreadable transcripts and as a render fallback. Treat as read-only. */ export const ZERO_COMPACTION_STATS: CompactionData = Object.freeze({ count: 0, @@ -73,37 +66,3 @@ export function accumulateCompactionStats(stats: CompactionData, record: unknown } } } - -/** - * Count context-compaction events and summarize their `compactMetadata` by - * scanning the transcript for `{type:'system', subtype:'compact_boundary'}` - * markers Claude Code writes on every compaction. Exact and immune to transient - * context-percentage noise. Sidechain (subagent) records are excluded. - * - * `trigger` missing or unrecognized is counted under `unknown` (never guessed). - * `tokensReclaimed` sums `preTokens - postTokens` only for markers where both - * are finite numbers; older markers without `postTokens` contribute 0. - */ -export function computeCompactionStats(lines: readonly string[]): CompactionData { - const stats = createCompactionStats(); - for (const line of lines) { - accumulateCompactionStats(stats, parseJsonlLine(line)); - } - return stats; -} - -/** Best-effort: returns zeroed stats when the transcript is missing or unreadable. */ -export async function getCompactionStats(transcriptPath: string): Promise { - try { - if (!fs.existsSync(transcriptPath)) { - return ZERO_COMPACTION_STATS; - } - const stats = createCompactionStats(); - for await (const line of iterateJsonlLines(transcriptPath)) { - accumulateCompactionStats(stats, parseJsonlLine(line)); - } - return stats; - } catch { - return ZERO_COMPACTION_STATS; - } -} diff --git a/src/utils/context-window.ts b/src/utils/context-window.ts index 1df09383..2ca94c42 100644 --- a/src/utils/context-window.ts +++ b/src/utils/context-window.ts @@ -20,21 +20,28 @@ function toFiniteNonNegativeNumber(value: unknown): number | null { return Math.max(0, value); } -interface CurrentUsageObject { +export interface UsageTokenObject { input_tokens?: number; output_tokens?: number; cache_creation_input_tokens?: number; cache_read_input_tokens?: number; } -interface CurrentUsageTokens { +export interface UsageTokens { input: number; output: number; creation: number; read: number; } -function parseCurrentUsageTokens(usage: CurrentUsageObject): CurrentUsageTokens { +/** + * Coerces one usage object to non-negative finite counts. + * + * @remarks + * Shared by the live status JSON and the transcript fallback so a malformed + * count cannot make the two paths disagree about the same session. + */ +export function parseUsageTokens(usage: UsageTokenObject): UsageTokens { return { input: toFiniteNonNegativeNumber(usage.input_tokens) ?? 0, output: toFiniteNonNegativeNumber(usage.output_tokens) ?? 0, @@ -43,6 +50,11 @@ function parseCurrentUsageTokens(usage: CurrentUsageObject): CurrentUsageTokens }; } +/** Tokens the model re-reads on the next turn: fresh input plus both cache halves. */ +export function contextLengthFromUsageTokens(tokens: UsageTokens): number { + return tokens.input + tokens.creation + tokens.read; +} + function clampPercentage(value: number): number { return Math.max(0, Math.min(100, value)); } @@ -77,10 +89,11 @@ export function getContextWindowMetrics(data?: StatusJSON): ContextWindowMetrics currentUsageTotalTokens = toFiniteNonNegativeNumber(contextWindow.current_usage); contextLengthTokens = currentUsageTotalTokens; } else if (contextWindow.current_usage && typeof contextWindow.current_usage === 'object') { - const { input, output, creation, read } = parseCurrentUsageTokens(contextWindow.current_usage); + const usageTokens = parseUsageTokens(contextWindow.current_usage); + const { input, output, creation, read } = usageTokens; currentUsageTotalTokens = input + output + creation + read; - contextLengthTokens = input + creation + read; + contextLengthTokens = contextLengthFromUsageTokens(usageTokens); cachedTokens = creation + read; } @@ -169,6 +182,6 @@ export function getContextWindowTurnCacheTokens(data?: StatusJSON): TurnCacheTok return null; } - const { input, creation, read } = parseCurrentUsageTokens(usage); + const { input, creation, read } = parseUsageTokens(usage); return { read, creation, input }; } diff --git a/src/utils/jsonl-lines.ts b/src/utils/jsonl-lines.ts index 183b1f6e..fc126c42 100644 --- a/src/utils/jsonl-lines.ts +++ b/src/utils/jsonl-lines.ts @@ -1,73 +1,8 @@ import * as fs from 'fs'; import { StringDecoder } from 'string_decoder'; -import { promisify } from 'util'; +/** Read size for both sync iterators. Exported so tests size records against it rather than a copy. */ export const JSONL_READ_CHUNK_BYTES = 1024 * 1024; -const stat = promisify(fs.stat); -const statSync = fs.statSync; - -/** Transcripts to retain. A render reads the session transcript and the subagent transcripts it references. */ -const MAX_CACHED_FILES = 8; - -interface CacheEntry { - /** Size and modification time, so an appended transcript is re-read rather than served stale. */ - readonly version: string; - readonly lines: readonly string[]; -} - -/** - * Split lines, keyed by file identity. - * - * @remarks - * Compatibility callers can still request a materialized line array. Reuse - * keeps repeated reads of an unchanged file to one walk, while transcript - * analysis uses the streaming iterators directly. - */ -const lineCache = new Map(); - -/** - * Identifies the file itself rather than the string used to reach it. - * - * @remarks - * Windows accepts many spellings of one path, and the code mixes them: a - * transcript path arrives with backslashes while a glob yields forward slashes. - * Keying on the device and inode gives those one entry instead of several. - * Inodes exceed the safe integer range, so the stat is taken as bigint. - */ -function identify(stats: fs.BigIntStats, filePath: string): string { - // A filesystem that reports no inode leaves the path as the only identity. - if (stats.ino === 0n) { - return `path:${filePath}`; - } - - return `ino:${stats.dev}:${stats.ino}`; -} - -function versionOf(stats: fs.BigIntStats): string { - return `${stats.size}:${stats.mtimeNs}`; -} - -function readCached(identity: string, version: string): readonly string[] | undefined { - const entry = lineCache.get(identity); - return entry?.version === version ? entry.lines : undefined; -} - -function writeCached(identity: string, version: string, lines: readonly string[]): readonly string[] { - // Re-inserting moves the entry to the end, so eviction stays least-recently-written. - lineCache.delete(identity); - lineCache.set(identity, { version, lines }); - - while (lineCache.size > MAX_CACHED_FILES) { - const oldest = lineCache.keys().next(); - if (oldest.done) { - break; - } - - lineCache.delete(oldest.value); - } - - return lines; -} /** * Splits byte chunks using JSONL's LF delimiter without interpreting Unicode @@ -283,73 +218,6 @@ export function* iterateJsonlLinesReverseSync(filePath: string): Generator { - const readLines = async (): Promise => { - const lines: string[] = []; - for await (const line of iterateJsonlLines(filePath)) { - lines.push(line); - } - return lines; - }; - - if (options?.cache === false) { - return readLines(); - } - - const stats = await stat(filePath, { bigint: true }); - const identity = identify(stats, filePath); - const version = versionOf(stats); - - const cached = readCached(identity, version); - if (cached !== undefined) { - return cached; - } - - return writeCached(identity, version, await readLines()); -} - -export function readJsonlLinesSync(filePath: string, options?: ReadJsonlLinesOptions): readonly string[] { - if (options?.cache === false) { - return Array.from(iterateJsonlLinesSync(filePath)); - } - - const stats = statSync(filePath, { bigint: true }); - const identity = identify(stats, filePath); - const version = versionOf(stats); - - const cached = readCached(identity, version); - if (cached !== undefined) { - return cached; - } - - return writeCached(identity, version, Array.from(iterateJsonlLinesSync(filePath))); -} - export function parseJsonlLine(line: string): unknown { try { return JSON.parse(line) as unknown; diff --git a/src/utils/jsonl-metrics.ts b/src/utils/jsonl-metrics.ts index 6e30b63a..8d733194 100644 --- a/src/utils/jsonl-metrics.ts +++ b/src/utils/jsonl-metrics.ts @@ -14,6 +14,11 @@ import { getCompactBoundaryPostTokens, isCompactBoundary } from './compaction'; +import { + contextLengthFromUsageTokens, + parseUsageTokens, + type UsageTokens +} from './context-window'; import { iterateJsonlLines, parseJsonlLine @@ -24,16 +29,6 @@ import { } from './jsonl-metadata'; import { getSessionNameFromRecord } from './jsonl-session'; -export interface SpeedMetricsOptions { - includeSubagents?: boolean; - windowSeconds?: number; -} - -export interface SpeedMetricsCollectionOptions { - includeSubagents?: boolean; - windowSeconds?: number[]; -} - export interface SpeedMetricsCollection { sessionAverage: SpeedMetrics; windowed: Record; @@ -86,15 +81,8 @@ interface CollectedSpeedMetrics { latestTimestampMs: number | null; } -interface RetainedTokenUsage { - inputTokens: number; - outputTokens: number; - cacheReadTokens: number; - cacheCreationTokens: number; -} - interface TokenMetricEntry { - usage: RetainedTokenUsage; + usage: UsageTokens; stopReason: string | null | undefined; timestampMs: number | null; isMainChain: boolean; @@ -105,9 +93,9 @@ interface TokenMetricAccumulator { outputTokens: number; cacheReadTokens: number; cacheCreationTokens: number; - mostRecentMainChainUsage: RetainedTokenUsage | null; + mostRecentMainChainUsage: UsageTokens | null; mostRecentTimestampMs: number | null; - mostRecentPostCompactionUsage: RetainedTokenUsage | null; + mostRecentPostCompactionUsage: UsageTokens | null; mostRecentPostCompactionTimestampMs: number | null; } @@ -156,10 +144,10 @@ function accumulateTokenMetricEntry( includePostCompactionUsage: boolean ): void { const { usage } = entry; - accumulator.inputTokens += usage.inputTokens; - accumulator.outputTokens += usage.outputTokens; - accumulator.cacheReadTokens += usage.cacheReadTokens; - accumulator.cacheCreationTokens += usage.cacheCreationTokens; + accumulator.inputTokens += usage.input; + accumulator.outputTokens += usage.output; + accumulator.cacheReadTokens += usage.read; + accumulator.cacheCreationTokens += usage.creation; if (!entry.isMainChain || entry.timestampMs === null) { return; @@ -201,12 +189,7 @@ function collectTokenMetricRecord(state: TokenMetricState, data: TranscriptLine const usage = message?.usage; if (usage) { const entry: TokenMetricEntry = { - usage: { - inputTokens: usage.input_tokens || 0, - outputTokens: usage.output_tokens || 0, - cacheReadTokens: usage.cache_read_input_tokens ?? 0, - cacheCreationTokens: usage.cache_creation_input_tokens ?? 0 - }, + usage: parseUsageTokens(usage), stopReason: message.stop_reason, timestampMs, isMainChain: data?.isSidechain !== true && !data?.isApiErrorMessage @@ -230,8 +213,8 @@ function finishTokenMetrics(state: TokenMetricState): TokenMetrics { accumulateTokenMetricEntry(state.metrics, state.lastUsageEntry, !state.boundaryAfterLastUsage); } - const contextLengthFromUsage = (usage: RetainedTokenUsage | null): number | null => usage - ? usage.inputTokens + usage.cacheReadTokens + usage.cacheCreationTokens + const contextLengthFromUsage = (usage: UsageTokens | null): number | null => usage + ? contextLengthFromUsageTokens(usage) : null; const contextLength = state.sawCompactBoundary ? (contextLengthFromUsage(state.metrics.mostRecentPostCompactionUsage) ?? state.lastCompactBoundaryPostTokens ?? 0) @@ -271,16 +254,6 @@ function collectAgentIds(value: unknown, agentIds: Set) { } } -export async function getSessionDuration(transcriptPath: string): Promise { - const result = await scanTranscript(transcriptPath, { includeSessionDuration: true }); - return result.sessionDuration; -} - -export async function getTokenMetrics(transcriptPath: string): Promise { - const result = await scanTranscript(transcriptPath, { includeTokenMetrics: true }); - return result.tokenMetrics ?? createEmptyTokenMetrics(); -} - function parseTimestampMs(value: string | undefined): number | null { if (!value) { return null; @@ -382,9 +355,10 @@ function collectSpeedMetricRecord( interval = { startMs: state.lastUserTimestampMs, endMs: timestampMs }; } + const usage = parseUsageTokens(data.message.usage); state.requests.push({ - inputTokens: data.message.usage.input_tokens || 0, - outputTokens: data.message.usage.output_tokens || 0, + inputTokens: usage.input, + outputTokens: usage.output, assistantTimestampMs: timestampMs, interval }); @@ -688,21 +662,6 @@ function getSubagentTranscriptPaths(transcriptPath: string, referencedAgentIds: return matchedPaths; } -export async function getSpeedMetricsCollection( - transcriptPath: string, - options: SpeedMetricsCollectionOptions = {} -): Promise { - const result = await scanTranscript(transcriptPath, { - includeSpeedMetrics: true, - includeSubagents: options.includeSubagents, - speedWindowSeconds: options.windowSeconds - }); - return result.speedMetricsCollection ?? { - sessionAverage: createEmptySpeedMetrics(), - windowed: buildEmptyWindowedMetrics(normalizeSpeedWindows(options.windowSeconds)) - }; -} - export async function getTranscriptAnalysis( transcriptPath: string, options: TranscriptAnalysisOptions = {} @@ -721,20 +680,3 @@ export async function getTranscriptAnalysis( sessionName: result.sessionName }; } - -export async function getSpeedMetrics( - transcriptPath: string, - options: SpeedMetricsOptions = {} -): Promise { - const requestedWindow = normalizeWindowSeconds(options.windowSeconds); - const metricsCollection = await getSpeedMetricsCollection(transcriptPath, { - includeSubagents: options.includeSubagents, - windowSeconds: requestedWindow ? [requestedWindow] : [] - }); - - if (requestedWindow === null) { - return metricsCollection.sessionAverage; - } - - return metricsCollection.windowed[requestedWindow.toString()] ?? createEmptySpeedMetrics(); -} diff --git a/src/utils/jsonl.ts b/src/utils/jsonl.ts index 1b071eb0..f862580c 100644 --- a/src/utils/jsonl.ts +++ b/src/utils/jsonl.ts @@ -5,13 +5,7 @@ export { writeBlockCache } from './jsonl-cache'; export { getBlockMetrics } from './jsonl-blocks'; -export { - getSessionDuration, - getSpeedMetrics, - getSpeedMetricsCollection, - getTokenMetrics, - getTranscriptAnalysis -} from './jsonl-metrics'; +export { getTranscriptAnalysis } from './jsonl-metrics'; export type { TranscriptAnalysis, TranscriptAnalysisOptions