From 987c1e103f8433e79c15214824166e1644d46a71 Mon Sep 17 00:00:00 2001 From: wellkilo Date: Sun, 13 Sep 2026 04:55:52 +0800 Subject: [PATCH] fix(metrics): bound incremental snapshot recovery Cap JSONL line buffering and per-hook catch-up work, persist discard cursors for oversized records, report retention failures, normalize malformed token totals, and strengthen bounded-read regression fixtures. --- scripts/hooks/cost-tracker.js | 16 +- scripts/hooks/ecc-metrics-bridge.js | 4 +- scripts/lib/session-cost-snapshot.js | 277 ++++++++++++++++-------- skills/cost-tracking/SKILL.md | 2 + tests/hooks/cost-tracker.test.js | 51 +++++ tests/hooks/ecc-metrics-bridge.test.js | 28 ++- tests/lib/session-cost-snapshot.test.js | 153 +++++++++++-- 7 files changed, 424 insertions(+), 107 deletions(-) diff --git a/scripts/hooks/cost-tracker.js b/scripts/hooks/cost-tracker.js index 63d61170d..cf36168b1 100755 --- a/scripts/hooks/cost-tracker.js +++ b/scripts/hooks/cost-tracker.js @@ -109,7 +109,17 @@ function isSonnet5(model) { function toNumber(v) { const n = Number(v); - return Number.isFinite(n) ? n : 0; + return Number.isFinite(n) && n >= 0 ? n : 0; +} + +function normalizeUsageTotals(totals) { + return { + inputTokens: toNumber(totals.inputTokens), + outputTokens: toNumber(totals.outputTokens), + cacheWriteTokens: toNumber(totals.cacheWriteTokens), + cacheReadTokens: toNumber(totals.cacheReadTokens), + model: totals.model + }; } /** @@ -167,7 +177,9 @@ function sumUsageFromTranscript(transcriptPath) { cacheReadTokens += toNumber(u.cache_read_input_tokens); } - return { inputTokens, outputTokens, cacheWriteTokens, cacheReadTokens, model }; + return normalizeUsageTotals({ + inputTokens, outputTokens, cacheWriteTokens, cacheReadTokens, model + }); } // 1MB, matching the other Stop hooks. The Stop payload carries diff --git a/scripts/hooks/ecc-metrics-bridge.js b/scripts/hooks/ecc-metrics-bridge.js index 8f10a3a49..31ecad948 100644 --- a/scripts/hooks/ecc-metrics-bridge.js +++ b/scripts/hooks/ecc-metrics-bridge.js @@ -155,7 +155,7 @@ function readSessionCost(sessionId) { 'malformed', costsPath, `${snapshotResult.malformed}:${snapshotResult.malformedSignature}`, - `[ecc-metrics-bridge] skipped ${snapshotResult.malformed} malformed line(s) in ${costsPath}\n` + `[ecc-metrics-bridge] skipped ${snapshotResult.malformed} malformed line(s) during the snapshot scan of ${costsPath}\n` ); } if (snapshotResult.invalid > 0) { @@ -163,7 +163,7 @@ function readSessionCost(sessionId) { 'invalid-row', costsPath, `${snapshotResult.invalid}:${snapshotResult.invalidSignature}`, - `[ecc-metrics-bridge] skipped ${snapshotResult.invalid} invalid cumulative row(s) for ${sessionId} in ${costsPath}\n` + `[ecc-metrics-bridge] skipped ${snapshotResult.invalid} invalid cumulative row(s) for ${sessionId} during the snapshot scan of ${costsPath}\n` ); } if (snapshotResult.snapshotError) { diff --git a/scripts/lib/session-cost-snapshot.js b/scripts/lib/session-cost-snapshot.js index 04a5e45b0..f95ddc8cb 100644 --- a/scripts/lib/session-cost-snapshot.js +++ b/scripts/lib/session-cost-snapshot.js @@ -11,6 +11,8 @@ const COST_SNAPSHOT_SCHEMA_VERSION = 'ecc.cost-snapshot.v1'; const COST_SNAPSHOT_DIRECTORY = 'cost-snapshots'; const COST_LOG_FILENAME = 'costs.jsonl'; const READ_CHUNK_BYTES = 64 * 1024; +const MAX_JSONL_LINE_BYTES = 1024 * 1024; +const MAX_SCAN_BYTES = 16 * 1024 * 1024; const FINGERPRINT_WINDOW_BYTES = 256; const PRUNE_INTERVAL_MS = 24 * 60 * 60 * 1000; const SNAPSHOT_MAX_AGE_MS = 30 * 24 * 60 * 60 * 1000; @@ -109,42 +111,121 @@ function validSnapshotBase(snapshot, descriptor, stat, sessionId) { source.fingerprint, fingerprintProcessedPrefix(descriptor, source.offset_bytes) )) return null; - return { row: snapshot.row, offset: source.offset_bytes }; + return { + row: snapshot.row, + offset: source.offset_bytes, + discardingLine: source.discarding_line === true + }; } -function scanJsonlRange(descriptor, start, end, sessionId, initialRow) { +function createScanState(initialRow) { + return { + latestRow: initialRow, + committedRow: initialRow, + malformed: 0, + invalid: 0, + malformedHasher: crypto.createHash('sha256'), + invalidHasher: crypto.createHash('sha256') + }; +} + +function processCostLine(state, line, sessionId, committed = true) { + if (!line.trim()) return state; + try { + const row = JSON.parse(line); + if (row.session_id !== sessionId) return state; + if (!isValidCostRow(row, sessionId)) { + if (!committed) return state; + return { + ...state, + invalid: state.invalid + 1, + invalidHasher: state.invalidHasher.copy().update(line).update('\0') + }; + } + return { + ...state, + latestRow: chooseNewerCumulativeRow(state.latestRow, row), + committedRow: committed + ? chooseNewerCumulativeRow(state.committedRow, row) + : state.committedRow + }; + } catch { + if (!committed) return state; + return { + ...state, + malformed: state.malformed + 1, + malformedHasher: state.malformedHasher.copy().update(line).update('\0') + }; + } +} + +function markOversizedLine(state, pendingChunks, segment) { + const hasher = state.malformedHasher.copy(); + for (const chunk of pendingChunks) hasher.update(chunk); + const remaining = Math.max(0, MAX_JSONL_LINE_BYTES - pendingChunks.reduce( + (total, chunk) => total + chunk.length, + 0 + )); + hasher.update(segment.subarray(0, remaining)).update('\0'); + return { ...state, malformed: state.malformed + 1, malformedHasher: hasher }; +} + +function consumeLineSegment(scan, segment, terminated, sessionId) { + if (scan.discardingLine) { + return { ...scan, discardingLine: !terminated }; + } + if (scan.pendingBytes + segment.length > MAX_JSONL_LINE_BYTES) { + return { + state: markOversizedLine(scan.state, scan.pendingChunks, segment), + pendingChunks: [], + pendingBytes: 0, + discardingLine: !terminated + }; + } + const pendingChunks = segment.length > 0 + ? [...scan.pendingChunks, Buffer.from(segment)] + : scan.pendingChunks; + const pendingBytes = scan.pendingBytes + segment.length; + if (!terminated) return { ...scan, pendingChunks, pendingBytes }; + const line = Buffer.concat(pendingChunks, pendingBytes).toString('utf8'); + return { + state: processCostLine(scan.state, line, sessionId), + pendingChunks: [], + pendingBytes: 0, + discardingLine: false + }; +} + +function consumeJsonlChunk(scan, chunk, sessionId, absoluteStart, processedOffset) { + let nextScan = scan; + let nextOffset = processedOffset; + let segmentStart = 0; + for (;;) { + const newlineIndex = chunk.indexOf(0x0a, segmentStart); + if (newlineIndex < 0) break; + nextScan = consumeLineSegment( + nextScan, chunk.subarray(segmentStart, newlineIndex), true, sessionId + ); + nextOffset = absoluteStart + newlineIndex + 1; + segmentStart = newlineIndex + 1; + } + nextScan = consumeLineSegment( + nextScan, chunk.subarray(segmentStart), false, sessionId + ); + if (nextScan.discardingLine) nextOffset = absoluteStart + chunk.length; + return { scan: nextScan, processedOffset: nextOffset }; +} + +function scanJsonlRange(descriptor, start, end, sessionId, initialRow, initialDiscard = false) { const buffer = Buffer.allocUnsafe(READ_CHUNK_BYTES); + let lineScan = { + state: createScanState(initialRow), + pendingChunks: [], + pendingBytes: 0, + discardingLine: initialDiscard + }; let position = start; let processedOffset = start; - let pending = Buffer.alloc(0); - let latestRow = initialRow; - let committedRow = initialRow; - let malformed = 0; - let invalid = 0; - const malformedHasher = crypto.createHash('sha256'); - const invalidHasher = crypto.createHash('sha256'); - - const processLine = (line, committed = true) => { - if (!line.trim()) return; - try { - const row = JSON.parse(line); - if (row.session_id !== sessionId) return; - if (!isValidCostRow(row, sessionId)) { - if (committed) { - invalid += 1; - invalidHasher.update(line).update('\0'); - } - return; - } - latestRow = chooseNewerCumulativeRow(latestRow, row); - if (committed) committedRow = chooseNewerCumulativeRow(committedRow, row); - } catch { - if (committed) { - malformed += 1; - malformedHasher.update(line).update('\0'); - } - } - }; while (position < end) { const bytesRead = fs.readSync( @@ -155,33 +236,40 @@ function scanJsonlRange(descriptor, start, end, sessionId, initialRow) { position ); if (bytesRead === 0) break; - const combined = pending.length > 0 - ? Buffer.concat([pending, buffer.subarray(0, bytesRead)]) - : buffer.subarray(0, bytesRead); - let lineStart = 0; - for (;;) { - const newlineIndex = combined.indexOf(0x0a, lineStart); - if (newlineIndex < 0) break; - processLine(combined.subarray(lineStart, newlineIndex).toString('utf8')); - lineStart = newlineIndex + 1; - } - pending = Buffer.from(combined.subarray(lineStart)); + const consumed = consumeJsonlChunk( + lineScan, buffer.subarray(0, bytesRead), sessionId, position, processedOffset + ); + lineScan = consumed.scan; + processedOffset = consumed.processedOffset; position += bytesRead; - processedOffset = position - pending.length; } - if (pending.toString('utf8').trim()) processLine(pending.toString('utf8'), false); + if (lineScan.pendingBytes > 0) { + const line = Buffer.concat(lineScan.pendingChunks, lineScan.pendingBytes).toString('utf8'); + lineScan = { + ...lineScan, + state: processCostLine(lineScan.state, line, sessionId, false) + }; + } + const { state } = lineScan; return { - row: latestRow, - committedRow, + row: state.latestRow, + committedRow: state.committedRow, processedOffset, - malformed, - invalid, - malformedSignature: malformed > 0 ? malformedHasher.digest('hex').slice(0, 16) : null, - invalidSignature: invalid > 0 ? invalidHasher.digest('hex').slice(0, 16) : null + malformed: state.malformed, + invalid: state.invalid, + malformedSignature: state.malformed > 0 + ? state.malformedHasher.digest('hex').slice(0, 16) + : null, + invalidSignature: state.invalid > 0 + ? state.invalidHasher.digest('hex').slice(0, 16) + : null, + discardingLine: lineScan.discardingLine }; } -function writeSnapshotAtOffset(metricsDir, sessionId, row, descriptor, stat, offset) { +function writeSnapshotAtOffset( + metricsDir, sessionId, row, descriptor, stat, offset, discardingLine = false +) { if (row !== null && !isValidCostRow(row, sessionId)) return false; const snapshot = { schema_version: COST_SNAPSHOT_SCHEMA_VERSION, @@ -189,6 +277,7 @@ function writeSnapshotAtOffset(metricsDir, sessionId, row, descriptor, stat, off identity: sourceIdentity(stat), offset_bytes: offset, mtime_ms: stat.mtimeMs, + discarding_line: discardingLine, fingerprint: fingerprintProcessedPrefix(descriptor, offset) }, row @@ -199,9 +288,6 @@ function writeSnapshotAtOffset(metricsDir, sessionId, row, descriptor, stat, off { beforeRename() { const current = fs.fstatSync(descriptor); - if (sourceIdentity(current) !== snapshot.source.identity) { - throw new Error('Cost log identity changed during snapshot publication'); - } if (current.size < offset) { throw new Error('Cost log was truncated during snapshot publication'); } @@ -220,6 +306,36 @@ function writeSnapshotAtOffset(metricsDir, sessionId, row, descriptor, stat, off return true; } +function emptySnapshotResult(row) { + return { + row, + scannedBytes: 0, + malformed: 0, + invalid: 0, + malformedSignature: null, + invalidSignature: null, + snapshotError: null + }; +} + +function publishScanSnapshot(metricsDir, sessionId, scan, descriptor, stat) { + if (!scan.committedRow && scan.processedOffset === 0) return null; + try { + writeSnapshotAtOffset( + metricsDir, + sessionId, + scan.committedRow, + descriptor, + stat, + scan.processedOffset, + scan.discardingLine + ); + return null; + } catch (error) { + return error; + } +} + function refreshSessionCostSnapshot(metricsDir, sessionId) { assertSafeSessionId(sessionId); const costsPath = path.join(metricsDir, COST_LOG_FILENAME); @@ -228,39 +344,19 @@ function refreshSessionCostSnapshot(metricsDir, sessionId) { const stat = fs.fstatSync(descriptor); const snapshot = readJsonFile(getCostSnapshotPath(metricsDir, sessionId)); const base = validSnapshotBase(snapshot, descriptor, stat, sessionId); - if (base?.offset === stat.size) { - return { - row: base.row, - scannedBytes: 0, - malformed: 0, - invalid: 0, - malformedSignature: null, - invalidSignature: null, - snapshotError: null - }; - } + if (base?.offset === stat.size) return emptySnapshotResult(base.row); + const scanEnd = Math.min(stat.size, (base?.offset || 0) + MAX_SCAN_BYTES); const scan = scanJsonlRange( descriptor, base?.offset || 0, - stat.size, + scanEnd, sessionId, - base?.row || null + base?.row || null, + base?.discardingLine || false + ); + const snapshotError = publishScanSnapshot( + metricsDir, sessionId, scan, descriptor, stat ); - let snapshotError = null; - if (scan.committedRow || scan.processedOffset > 0) { - try { - writeSnapshotAtOffset( - metricsDir, - sessionId, - scan.committedRow, - descriptor, - stat, - scan.processedOffset - ); - } catch (error) { - snapshotError = error; - } - } return { row: scan.row, scannedBytes: scan.processedOffset - (base?.offset || 0), @@ -308,8 +404,10 @@ function appendSessionCostRow(metricsDir, sessionId, row) { if (result.snapshotError) throw result.snapshotError; try { maybePruneSessionCostSnapshots(metricsDir); - } catch { - // Retention is opportunistic and retried by a later update. + } catch (error) { + // Retention is opportunistic and retried by a later update, but a + // persistent failure remains visible without rolling back the log append. + warnSessionCostSnapshotFailure('retention', metricsDir, sessionId, error); } return JSON.stringify(result.row) === JSON.stringify(row); } @@ -341,10 +439,10 @@ function maybePruneSessionCostSnapshots(metricsDir, options = {}) { try { const intervalIsFresh = now - fs.statSync(markerPath).mtimeMs < PRUNE_INTERVAL_MS; if (intervalIsFresh && snapshotEntries.length <= maxSnapshots) return 0; - } catch { /* missing marker */ } + } catch (error) { + if (error.code !== 'ENOENT') throw error; + } } - fs.writeFileSync(markerPath, String(now), { encoding: 'utf8', mode: 0o600 }); - const snapshots = snapshotEntries .map(entry => { const filePath = path.join(snapshotDir, entry.name); @@ -361,8 +459,11 @@ function maybePruneSessionCostSnapshots(metricsDir, options = {}) { fs.rmSync(entry.filePath, { force: true }); removed += 1; } - } catch { /* already replaced or removed */ } + } catch (error) { + if (error.code !== 'ENOENT') throw error; + } } + fs.writeFileSync(markerPath, String(now), { encoding: 'utf8', mode: 0o600 }); return removed; } diff --git a/skills/cost-tracking/SKILL.md b/skills/cost-tracking/SKILL.md index 4347ce8f4..d21a401b4 100644 --- a/skills/cost-tracking/SKILL.md +++ b/skills/cost-tracking/SKILL.md @@ -23,6 +23,8 @@ session total without rescanning all history. Treat those files as a rebuildable cache; each snapshot stores a byte cursor so only newly appended rows are scanned. Stable reads are O(1), while updates are O(new bytes). Stale entries are pruned after 30 days or when the directory exceeds 512 sessions. +Cold catch-up work is limited to 16 MiB per hook invocation, and malformed +unterminated rows larger than 1 MiB are discarded with a resumable cursor. Reports and exports should continue to use `costs.jsonl`. Row schema: diff --git a/tests/hooks/cost-tracker.test.js b/tests/hooks/cost-tracker.test.js index 066c3226d..e1152116b 100644 --- a/tests/hooks/cost-tracker.test.js +++ b/tests/hooks/cost-tracker.test.js @@ -249,6 +249,57 @@ function runTests() { fs.rmSync(tmpHome, { recursive: true, force: true }); }) ? passed++ : failed++); + (test('normalizes malformed negative and non-finite transcript usage', () => { + const tmpHome = makeTempDir(); + const transcriptPath = path.join(tmpHome, 'session.jsonl'); + writeTranscript(transcriptPath, [{ + type: 'assistant', + message: { + id: 'msg_invalid_usage', + model: 'claude-sonnet-4-20250514', + usage: { + input_tokens: -100, + output_tokens: 'Infinity', + cache_creation_input_tokens: -20, + cache_read_input_tokens: 'not-a-number', + }, + }, + }, { + type: 'assistant', + message: { + id: 'msg_overflow_1', + model: 'claude-sonnet-4-20250514', + usage: { input_tokens: 1e308, output_tokens: 0 }, + }, + }, { + type: 'assistant', + message: { + id: 'msg_overflow_2', + model: 'claude-sonnet-4-20250514', + usage: { input_tokens: 1e308, output_tokens: 0 }, + }, + }]); + + const result = runScript( + { session_id: 'invalid-usage', transcript_path: transcriptPath }, + withTempHome(tmpHome) + ); + assert.strictEqual(result.code, 0, result.stderr); + const metricsFile = path.join(tmpHome, '.claude', 'metrics', 'costs.jsonl'); + const recorded = JSON.parse(fs.readFileSync(metricsFile, 'utf8').trim()); + assert.deepStrictEqual( + { + input: recorded.input_tokens, + output: recorded.output_tokens, + cacheWrite: recorded.cache_write_tokens, + cacheRead: recorded.cache_read_tokens, + cost: recorded.estimated_cost_usd, + }, + { input: 0, output: 0, cacheWrite: 0, cacheRead: 0, cost: 0 } + ); + fs.rmSync(tmpHome, { recursive: true, force: true }); + }) ? passed++ : failed++); + // 3. Handles empty input gracefully (test('handles empty input gracefully', () => { const tmpHome = makeTempDir(); diff --git a/tests/hooks/ecc-metrics-bridge.test.js b/tests/hooks/ecc-metrics-bridge.test.js index 246301622..c8dafe6dc 100644 --- a/tests/hooks/ecc-metrics-bridge.test.js +++ b/tests/hooks/ecc-metrics-bridge.test.js @@ -256,6 +256,18 @@ function runTests() { input_tokens: 750, output_tokens: 375 }; + const historicalRow = JSON.stringify({ + session_id: 'HISTORY', + estimated_cost_usd: 0, + input_tokens: 0, + output_tokens: 0 + }); + fs.writeFileSync( + path.join(metricsDir, 'costs.jsonl'), + `${historicalRow}\n`.repeat(100), + 'utf8' + ); + assert.ok(fs.statSync(path.join(metricsDir, 'costs.jsonl')).size > 3 * 256); appendSessionCostRow(metricsDir, 'S1', snapshotRow); fs.readSync = function measuredRead(descriptor, buffer, offset, length, position) { @@ -295,6 +307,18 @@ function runTests() { input_tokens: 100, output_tokens: 50 }; + const historicalRow = JSON.stringify({ + session_id: 'HISTORY', + estimated_cost_usd: 0, + input_tokens: 0, + output_tokens: 0 + }); + fs.writeFileSync( + path.join(metricsDir, 'costs.jsonl'), + `${historicalRow}\n`.repeat(200), + 'utf8' + ); + assert.ok(fs.statSync(path.join(metricsDir, 'costs.jsonl')).size > 3 * 1024); appendSessionCostRow(metricsDir, 'S1', first); appendSessionCostRow(metricsDir, 'S2', { session_id: 'S2', @@ -310,7 +334,7 @@ function runTests() { }; try { assert.deepStrictEqual(readSessionCost('S1'), { totalCost: 1, totalIn: 100, totalOut: 50 }); - assert.ok(bytesReadFromCostLog <= 2 * 1024); + assert.ok(bytesReadFromCostLog <= 3 * 1024); } finally { fs.readSync = originalReadSync; } @@ -448,6 +472,7 @@ function runTests() { totalOut: 100 }); assert.match(captured, /skipped 3 invalid cumulative row\(s\) for S1/); + assert.match(captured, /during the snapshot scan of/); } finally { process.stderr.write = originalStderrWrite; if (originalHome === undefined) delete process.env.HOME; @@ -541,6 +566,7 @@ function runTests() { const matches = captured.match(/skipped 2 malformed line\(s\)/g) || []; assert.strictEqual(matches.length, 1, `expected one aggregated malformed-line breadcrumb on stderr, got: ${captured}`); + assert.match(captured, /during the snapshot scan of/); } finally { process.stderr.write = originalStderrWrite; if (originalHome === undefined) delete process.env.HOME; diff --git a/tests/lib/session-cost-snapshot.test.js b/tests/lib/session-cost-snapshot.test.js index 153c2e932..5e004b49a 100644 --- a/tests/lib/session-cost-snapshot.test.js +++ b/tests/lib/session-cost-snapshot.test.js @@ -63,9 +63,11 @@ try { })) passed++; else failed++; if (test('a delayed older writer cannot lower the latest cumulative total', () => { - const older = row('session-order', 1); const newer = row('session-order', 2); - older.timestamp = new Date(Date.parse(newer.timestamp) + 1000).toISOString(); + const older = { + ...row('session-order', 1), + timestamp: new Date(Date.parse(newer.timestamp) + 1000).toISOString() + }; assert.strictEqual(appendSessionCostRow(root, 'session-order', newer), true); assert.strictEqual(appendSessionCostRow(root, 'session-order', older), false); assert.deepStrictEqual(readSessionCostSnapshot(root, 'session-order').row, newer); @@ -73,9 +75,11 @@ try { if (test('publication failure followed by an older writer still converges to the newer row', () => { const sessionId = 'session-publication-race'; - const older = row(sessionId, 1); const newer = row(sessionId, 2); - older.timestamp = new Date(Date.parse(newer.timestamp) + 1000).toISOString(); + const older = { + ...row(sessionId, 1), + timestamp: new Date(Date.parse(newer.timestamp) + 1000).toISOString() + }; const snapshotPath = getCostSnapshotPath(root, sessionId); const originalRenameSync = fs.renameSync; let injectedFailure = false; @@ -99,22 +103,25 @@ try { if (test('accepts increasing cumulative totals that share a timestamp', () => { const first = row('session-same-time', 1); - const next = row('session-same-time', 2); - next.timestamp = first.timestamp; + const next = { ...row('session-same-time', 2), timestamp: first.timestamp }; appendSessionCostRow(root, 'session-same-time', first); assert.strictEqual(appendSessionCostRow(root, 'session-same-time', next), true); assert.deepStrictEqual(readSessionCostSnapshot(root, 'session-same-time').row, next); })) passed++; else failed++; if (test('uses timestamps when cumulative dimensions move in opposite directions', () => { - const newer = row('session-mixed', 1); - newer.input_tokens = 200; - newer.output_tokens = 100; - newer.timestamp = '2026-01-02T00:00:00.000Z'; - const delayedOlder = row('session-mixed', 2); - delayedOlder.input_tokens = 100; - delayedOlder.output_tokens = 50; - delayedOlder.timestamp = '2026-01-01T00:00:00.000Z'; + const newer = { + ...row('session-mixed', 1), + input_tokens: 200, + output_tokens: 100, + timestamp: '2026-01-02T00:00:00.000Z' + }; + const delayedOlder = { + ...row('session-mixed', 2), + input_tokens: 100, + output_tokens: 50, + timestamp: '2026-01-01T00:00:00.000Z' + }; appendSessionCostRow(root, 'session-mixed', newer); assert.strictEqual(appendSessionCostRow(root, 'session-mixed', delayedOlder), false); assert.deepStrictEqual(readSessionCostSnapshot(root, 'session-mixed').row, newer); @@ -189,6 +196,45 @@ try { assert.deepStrictEqual(refreshSessionCostSnapshot(caseRoot, 'utf8').row, current); })) passed++; else failed++; + if (test('bounds oversized unterminated rows and caches the discarded prefix', () => { + const caseRoot = path.join(root, 'oversized-line'); + fs.mkdirSync(caseRoot, { recursive: true }); + const oversizedBytes = 32 * 1024 * 1024; + fs.writeFileSync( + path.join(caseRoot, 'costs.jsonl'), + Buffer.alloc(oversizedBytes, 0x78) + ); + const originalConcat = Buffer.concat; + let copiedBytes = 0; + Buffer.concat = function measuredConcat(list, totalLength) { + copiedBytes += totalLength ?? list.reduce((sum, item) => sum + item.length, 0); + return originalConcat.call(this, list, totalLength); + }; + try { + const first = refreshSessionCostSnapshot(caseRoot, 'oversized'); + assert.strictEqual(first.row, null); + assert.strictEqual(first.malformed, 1); + assert.ok(first.scannedBytes > 0 && first.scannedBytes < oversizedBytes); + assert.ok(copiedBytes <= 2 * 1024 * 1024, `copied ${copiedBytes} bytes`); + const second = refreshSessionCostSnapshot(caseRoot, 'oversized'); + assert.strictEqual(second.scannedBytes, oversizedBytes - first.scannedBytes); + assert.strictEqual(second.malformed, 0); + const stable = refreshSessionCostSnapshot(caseRoot, 'oversized'); + assert.strictEqual(stable.scannedBytes, 0); + const recovered = row('oversized', 3); + fs.appendFileSync( + path.join(caseRoot, 'costs.jsonl'), + `\n${JSON.stringify(recovered)}\n`, + 'utf8' + ); + const resumed = refreshSessionCostSnapshot(caseRoot, 'oversized'); + assert.deepStrictEqual(resumed.row, recovered); + assert.ok(resumed.scannedBytes < 1024); + } finally { + Buffer.concat = originalConcat; + } + })) passed++; else failed++; + if (test('rebuilds after an in-place rewrite or inode rotation', () => { const caseRoot = path.join(root, 'rotation'); fs.mkdirSync(caseRoot, { recursive: true }); @@ -296,6 +342,85 @@ try { assert.strictEqual(remaining.length, 2); })) passed++; else failed++; + if (test('surfaces retention removal failures for the caller to report', () => { + const caseRoot = path.join(root, 'retention-failure'); + fs.mkdirSync(caseRoot, { recursive: true }); + const sessionId = 'retention-target'; + appendSessionCostRow(caseRoot, sessionId, row(sessionId, 1)); + const snapshotPath = getCostSnapshotPath(caseRoot, sessionId); + const markerPath = path.join(caseRoot, 'cost-snapshots', '.last-prune'); + fs.rmSync(markerPath, { force: true }); + const old = new Date(Date.now() - 10_000); + fs.utimesSync(snapshotPath, old, old); + const originalRmSync = fs.rmSync; + fs.rmSync = function failSnapshotRemoval(filePath, options) { + if (path.resolve(filePath) === path.resolve(snapshotPath)) { + const error = new Error('injected retention failure'); + error.code = 'EACCES'; + throw error; + } + return originalRmSync.call(this, filePath, options); + }; + try { + assert.throws( + () => maybePruneSessionCostSnapshots(caseRoot, { + force: true, + now: Date.now(), + maxAgeMs: 1 + }), + /injected retention failure/ + ); + assert.strictEqual( + fs.existsSync(markerPath), + false, + 'failed pruning must not defer the next retry' + ); + } finally { + fs.rmSync = originalRmSync; + } + })) passed++; else failed++; + + if (test('reports retention failures without rolling back the appended row', () => { + const caseRoot = path.join(root, 'retention-warning'); + fs.mkdirSync(caseRoot, { recursive: true }); + const snapshotDir = path.join(caseRoot, 'cost-snapshots'); + const originalReaddirSync = fs.readdirSync; + const originalWrite = process.stderr.write.bind(process.stderr); + let captured = ''; + fs.readdirSync = function failRetentionRead(directory, options) { + if (path.resolve(directory) === path.resolve(snapshotDir)) { + const error = new Error('injected retention read failure'); + error.code = 'EACCES'; + throw error; + } + return originalReaddirSync.call(this, directory, options); + }; + process.stderr.write = chunk => { + captured += String(chunk); + return true; + }; + try { + const current = row('retention-warning-session', 1); + assert.strictEqual(appendSessionCostRow( + caseRoot, + 'retention-warning-session', + current + ), true); + assert.strictEqual(appendSessionCostRow( + caseRoot, + 'retention-warning-session', + row('retention-warning-session', 2) + ), true); + const warnings = captured.match(/retention failed/g) || []; + assert.strictEqual(warnings.length, 1); + const persisted = fs.readFileSync(path.join(caseRoot, 'costs.jsonl'), 'utf8'); + assert.match(persisted, /retention-warning-session/); + } finally { + fs.readdirSync = originalReaddirSync; + process.stderr.write = originalWrite; + } + })) passed++; else failed++; + if (test('deduplicates snapshot warnings independently by failure kind', () => { const caseRoot = path.join(root, 'warning-dedupe'); const originalWrite = process.stderr.write.bind(process.stderr);