From 8326aee4a6225c8ac37753f81a1e2a8841359e1a Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Sat, 18 Jul 2026 16:26:12 -0700 Subject: [PATCH] perf(parser): parse appended session files from the cached offset instead of byte 0 --- src/parser.ts | 115 +++++++++++- src/session-cache.ts | 4 + tests/parser-incremental-append.test.ts | 238 ++++++++++++++++++++++++ tests/session-cache.test.ts | 11 ++ 4 files changed, 366 insertions(+), 2 deletions(-) create mode 100644 tests/parser-incremental-append.test.ts diff --git a/src/parser.ts b/src/parser.ts index 8f6818ac..bbfc65b3 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -1593,7 +1593,7 @@ async function scanProjectDirs( type FileInfo = { dirName: string; fp: NonNullable>>; source?: SessionSourceMetadata } const unchangedFiles: Array<{ filePath: string; dirName: string; source?: SessionSourceMetadata; cached: CachedFile }> = [] - const changedFiles: Array<{ filePath: string; info: FileInfo }> = [] + const changedFiles: Array<{ filePath: string; info: FileInfo; append?: { cached: CachedFile; readFromOffset: number } }> = [] const discoverProgress = createScanProgress('scanning claude project dirs', dirs.length) let dirsDone = 0 @@ -1607,6 +1607,12 @@ async function scanProjectDirs( const action = reconcileFile(fp, section.files[filePath]) if (action.action === 'unchanged') { unchangedFiles.push({ filePath, dirName, source, cached: section.files[filePath]! }) + } else if (action.action === 'appended') { + changedFiles.push({ + filePath, + info: { dirName, fp, source }, + append: { cached: section.files[filePath]!, readFromOffset: action.readFromOffset }, + }) } else { changedFiles.push({ filePath, info: { dirName, fp, source } }) } @@ -1629,10 +1635,78 @@ async function scanProjectDirs( const progressTotal = changedFiles.length let filesDone = 0 emitScanProgress({ kind: 'tick', provider: 'claude', done: 0, total: progressTotal }) - for (const { filePath, info } of changedFiles) { + for (const { filePath, info, append } of changedFiles) { delete section.files[filePath] try { + if (append) { + // Append-only growth: parse ONLY the bytes past the cached resume offset + // and merge with the cached turns, rather than re-reading the file from 0. + // On a studio machine where live agents constantly append to session + // JSONL, this is the dominant warm-run cost. The merged result is + // byte-for-byte identical to a full re-parse (see mergeBoundaryCalls). + const tracker = { lastCompleteLineOffset: append.readFromOffset } + const newEntries = await parseClaudeEntries(filePath, tracker, append.readFromOffset) + const cached = append.cached + + const newTurns = newEntries + ? groupIntoTurns(dedupeStreamingMessageIds(newEntries), seenMsgIds).map(parsedTurnToCachedTurn) + : [] + + const mergedTurns: CachedTurn[] = cached.turns.map(t => ({ ...t, calls: [...t.calls] })) + if (newTurns.length > 0) { + let startIdx = 0 + // A first new turn with no leading user message is a continuation of + // the last cached turn — merge its calls in (a full re-parse would put + // them in that same turn), then append the remaining new turns. + if (!newTurns[0]!.userMessage.trim() && mergedTurns.length > 0) { + const last = mergedTurns[mergedTurns.length - 1]! + last.calls = mergeBoundaryCalls(last.calls, newTurns[0]!.calls) + startIdx = 1 + } + for (let i = startIdx; i < newTurns.length; i++) mergedTurns.push(newTurns[i]!) + } + + // The cached region's dedup keys were not added to seenMsgIds (only + // unchanged files pre-seed it), so add them now — a full re-parse would + // have, and later files dedup cross-file against them. + for (const t of cached.turns) for (const c of t.calls) seenMsgIds.add(c.deduplicationKey) + + // First-cwd wins, and the first cwd lives in the cached region whenever + // one was resolved there; only re-derive if the cached region had none. + let canonicalCwd = cached.canonicalCwd + let canonicalProjectName = cached.canonicalProjectName + if (canonicalCwd === undefined && newEntries) { + const cwd = extractCanonicalCwd(newEntries) + const canonical = (cwd && !isCoworkSession(cwd, filePath)) ? await resolveCanonicalProjectPath(cwd) : undefined + canonicalCwd = canonical?.path + canonicalProjectName = canonical?.isWorktree ? projectNameFromPath(canonical.path, info.dirName) : undefined + } + + // Inventory is a sorted set union; cached (older entries) ∪ new = full. + const mcpInventory = newEntries + ? Array.from(new Set([...cached.mcpInventory, ...extractMcpInventory(newEntries)])).sort() + : cached.mcpInventory + + section.files[filePath] = { + fingerprint: info.fp, + lastCompleteLineOffset: tracker.lastCompleteLineOffset, + canonicalCwd, + canonicalProjectName, + mcpInventory, + turns: mergedTurns, + agentType: cached.agentType, + } + ;(diskCache as { _dirty?: boolean })._dirty = true + filesDone++ + await parseProgress.tick(filesDone) + if (filesDone % 50 === 0 || filesDone === progressTotal) { + emitScanProgress({ kind: 'tick', provider: 'claude', done: filesDone, total: progressTotal }) + } + if (onFileParsed) await onFileParsed() + continue + } + const tracker = { lastCompleteLineOffset: 0 } const entries = await parseClaudeEntries(filePath, tracker) if (!entries) { filesDone++; await parseProgress.tick(filesDone); continue } @@ -1968,15 +2042,52 @@ function cachedTurnToClassified(turn: CachedTurn): ClassifiedTurn { // ── Cache-Aware Parsing Helpers ──────────────────────────────────────── +// Merge the calls of the last cached turn with the calls parsed from the +// appended region when the appended region continues that turn (its first new +// content had no leading user message). This mirrors `dedupeStreamingMessageIds` +// at the call level: a Claude message re-emitted across the append boundary +// (same `msg.id`, or the trailing not-yet-newline-terminated line re-read from +// the resume offset) collapses to its LAST occurrence, keeping the FIRST +// occurrence's timestamp — byte-for-byte what a full re-parse of the combined +// stream produces. Synthetic `claude:` keys (id-less entries) are never +// collapsed, matching `getMessageId` returning null for them. +function mergeBoundaryCalls(cachedCalls: CachedCall[], newCalls: CachedCall[]): CachedCall[] { + const combined = [...cachedCalls, ...newCalls] + const firstIdx = new Map() + const lastIdx = new Map() + for (let i = 0; i < combined.length; i++) { + const key = combined[i]!.deduplicationKey + if (key.startsWith('claude:')) continue + if (!firstIdx.has(key)) firstIdx.set(key, i) + lastIdx.set(key, i) + } + if (lastIdx.size === 0) return combined + const result: CachedCall[] = [] + for (let i = 0; i < combined.length; i++) { + const call = combined[i]! + const key = call.deduplicationKey + if (key.startsWith('claude:')) { result.push(call); continue } + if (lastIdx.get(key) !== i) continue + if (firstIdx.get(key) !== i) { + result.push({ ...call, timestamp: combined[firstIdx.get(key)!]!.timestamp }) + continue + } + result.push(call) + } + return result +} + async function parseClaudeEntries( filePath: string, tracker: { lastCompleteLineOffset: number }, + startByteOffset?: number, ): Promise { const entries: JournalEntry[] = [] let hasLines = false for await (const line of readSessionLines(filePath, undefined, { largeLineAsBuffer: true, byteOffsetTracker: tracker, + ...(startByteOffset !== undefined ? { startByteOffset } : {}), })) { hasLines = true const entry = parseJsonlLine(line) diff --git a/src/session-cache.ts b/src/session-cache.ts index 9f8dc3f3..56b09421 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -435,6 +435,10 @@ export function reconcileFile( if ( cached.lastCompleteLineOffset !== undefined && + // Defensive: never resume past the file's current end. A truncate-then-regrow + // can leave the cached offset stranded beyond live bytes; reading from there + // would silently drop the appended tail, so fall back to a full re-parse. + cached.lastCompleteLineOffset <= current.sizeBytes && fp.dev === current.dev && fp.ino === current.ino && current.sizeBytes > fp.sizeBytes diff --git a/tests/parser-incremental-append.test.ts b/tests/parser-incremental-append.test.ts new file mode 100644 index 00000000..74812bcb --- /dev/null +++ b/tests/parser-incremental-append.test.ts @@ -0,0 +1,238 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' +import { mkdtemp, mkdir, writeFile, appendFile, readFile, rm, stat, unlink } from 'fs/promises' +import { join } from 'path' +import { tmpdir } from 'os' + +// Spy on readSessionLines so a test can PROVE the incremental path activates: +// only the appended-parse path passes a non-zero `startByteOffset`. The real +// implementation is preserved; we merely record the offset each call receives. +const readLineCalls: Array<{ filePath: string; startByteOffset?: number }> = [] +vi.mock('../src/fs-utils.js', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + readSessionLines: (filePath: string, skip?: unknown, options?: { startByteOffset?: number }) => { + readLineCalls.push({ filePath, startByteOffset: options?.startByteOffset }) + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return (actual.readSessionLines as any)(filePath, skip, options) + }, + } +}) + +import { parseAllSessions, clearSessionCache } from '../src/parser.js' +import { sessionCachePath } from '../src/session-cache.js' +import type { ProjectSummary } from '../src/types.js' + +let tmpDir: string +let projectDir: string +let sessionPath: string +const CWD = '/tmp/incr-proj' + +beforeEach(async () => { + clearSessionCache() + readLineCalls.length = 0 + tmpDir = await mkdtemp(join(tmpdir(), 'incr-append-')) + projectDir = join(tmpDir, 'projects', 'incr-proj') + await mkdir(projectDir, { recursive: true }) + sessionPath = join(projectDir, 'sess-1.jsonl') + process.env['CLAUDE_CONFIG_DIR'] = tmpDir + process.env['CODEBURN_DESKTOP_SESSIONS_DIR'] = join(tmpDir, 'desktop-sessions') +}) + +afterEach(async () => { + clearSessionCache() + delete process.env['CODEBURN_CACHE_DIR'] + await rm(tmpDir, { recursive: true, force: true }) +}) + +// ── fixture builders ─────────────────────────────────────────────────── + +function userLine(ts: string, text: string): string { + return JSON.stringify({ + type: 'user', sessionId: 'sess-1', timestamp: ts, cwd: CWD, + message: { role: 'user', content: text }, + }) +} + +function mcpLine(ts: string, addedNames: string[]): string { + return JSON.stringify({ + type: 'user', sessionId: 'sess-1', timestamp: ts, cwd: CWD, + attachment: { type: 'deferred_tools_delta', addedNames }, + }) +} + +function asstLine( + id: string, + ts: string, + usage: Record, + blocks: Array> = [], + model = 'claude-sonnet-4-5', +): string { + return JSON.stringify({ + type: 'assistant', sessionId: 'sess-1', timestamp: ts, cwd: CWD, + message: { id, type: 'message', role: 'assistant', model, content: blocks, usage }, + }) +} + +const readBlock = (file: string) => ({ type: 'tool_use', name: 'Read', input: { file_path: file } }) +const bashBlock = (cmd: string) => ({ type: 'tool_use', name: 'Bash', input: { command: cmd } }) + +// A representative multi-turn session: MCP inventory, tools, bash, and a +// streaming re-emit of one assistant message (same id, updated usage) inside a +// turn — exercises dedup, breakdowns, and turn assembly. +function baseLines(): string[] { + return [ + mcpLine('2026-05-01T10:00:00.000Z', ['mcp__ctx__search', 'mcp__ctx__fetch']), + userLine('2026-05-01T10:00:01.000Z', 'first task please'), + asstLine('msg-a', '2026-05-01T10:00:02.000Z', { input_tokens: 100, output_tokens: 20 }, [readBlock('/a.ts')]), + // streaming re-emit of msg-a with grown usage (last one wins) + asstLine('msg-a', '2026-05-01T10:00:03.000Z', { input_tokens: 100, output_tokens: 55, cache_read_input_tokens: 300 }, [readBlock('/a.ts')]), + userLine('2026-05-01T10:05:00.000Z', 'second task please'), + asstLine('msg-b', '2026-05-01T10:05:02.000Z', { input_tokens: 200, output_tokens: 80 }, [bashBlock('ls -la')]), + ] +} + +// ── helpers ──────────────────────────────────────────────────────────── + +async function parseWith(cacheDir: string): Promise { + clearSessionCache() + process.env['CODEBURN_CACHE_DIR'] = cacheDir + return parseAllSessions() +} + +// A cold full re-parse of the file's CURRENT contents, using a pristine cache +// dir so nothing is served incrementally — the correctness oracle. +async function coldFullReparse(): Promise { + const freshCache = await mkdtemp(join(tmpdir(), 'incr-cold-')) + try { + return await parseWith(freshCache) + } finally { + await rm(freshCache, { recursive: true, force: true }) + } +} + +function offsetsFor(path: string): Array { + return readLineCalls.filter(c => c.filePath === path).map(c => c.startByteOffset) +} + +// ── tests ────────────────────────────────────────────────────────────── + +describe('incremental append parsing', () => { + it('CORE: warm append merge deep-equals a cold full re-parse (with torn final line)', async () => { + const warmCache = await mkdtemp(join(tmpdir(), 'incr-warm-')) + + // 1) cold parse of the base file → warm cache seeded with offset+turns. + await writeFile(sessionPath, baseLines().join('\n') + '\n') + await parseWith(warmCache) + + const cachedOffset: number = JSON.parse(await readFile(sessionCachePath(), 'utf-8')) + .providers.claude.files[sessionPath].lastCompleteLineOffset + expect(cachedOffset).toBeGreaterThan(0) + + // 2) append a new complete turn plus a torn (invalid JSON, no newline) tail. + const appended = + userLine('2026-05-01T11:00:00.000Z', 'third task please') + '\n' + + asstLine('msg-c', '2026-05-01T11:00:02.000Z', { input_tokens: 300, output_tokens: 90 }, [readBlock('/c.ts'), bashBlock('grep x')]) + '\n' + + '{"type":"assistant","sessionId":"sess-1","timestamp":"2026-05-01T11:05' // torn: invalid + no newline + await appendFile(sessionPath, appended) + + // 3) warm parse (same cache) → must take the incremental path. + readLineCalls.length = 0 + const warm = await parseWith(warmCache) + + // proof: the session file was read from the cached offset, not byte 0. + expect(offsetsFor(sessionPath)).toContain(cachedOffset) + + // 4) oracle: cold full re-parse of the identical file. + const cold = await coldFullReparse() + + expect(warm).toEqual(cold) + await rm(warmCache, { recursive: true, force: true }) + }) + + it('EDGE: append after a previously-torn line completes still equals cold', async () => { + const warmCache = await mkdtemp(join(tmpdir(), 'incr-warm2-')) + + // Base file whose LAST line is a complete assistant entry WITHOUT a trailing + // newline — cached as a turn, but the resume offset sits before it. + const tail = asstLine('msg-b', '2026-05-01T10:05:02.000Z', { input_tokens: 200, output_tokens: 80 }, [bashBlock('ls -la')]) + const head = baseLines().slice(0, 5).join('\n') + '\n' + await writeFile(sessionPath, head + tail) // no trailing newline + await parseWith(warmCache) + + // Complete the boundary (newline terminates the former tail) and append more. + // msg-b is re-read from the offset -> must dedup against the cached copy. + const more = '\n' + + asstLine('msg-c', '2026-05-01T11:00:02.000Z', { input_tokens: 300, output_tokens: 90 }, [readBlock('/c.ts')]) + '\n' + await appendFile(sessionPath, more) + + const warm = await parseWith(warmCache) + const cold = await coldFullReparse() + expect(warm).toEqual(cold) + await rm(warmCache, { recursive: true, force: true }) + }) + + it('EDGE: only a torn partial appended (no new complete line) equals cold', async () => { + const warmCache = await mkdtemp(join(tmpdir(), 'incr-warm3-')) + await writeFile(sessionPath, baseLines().join('\n') + '\n') + await parseWith(warmCache) + + // Append only an incomplete line (no newline) — nothing new to commit yet. + await appendFile(sessionPath, '{"type":"assistant","partial":true') + const warm = await parseWith(warmCache) + const cold = await coldFullReparse() + expect(warm).toEqual(cold) + await rm(warmCache, { recursive: true, force: true }) + }) + + it('EDGE: file replaced (inode change) falls back to a full re-parse', async () => { + const warmCache = await mkdtemp(join(tmpdir(), 'incr-warm4-')) + await writeFile(sessionPath, baseLines().join('\n') + '\n') + await parseWith(warmCache) + const inoBefore = (await stat(sessionPath)).ino + + // Replace the file (new inode) with different, LARGER content. + await unlink(sessionPath) + const replaced = [ + ...baseLines(), + userLine('2026-05-01T12:00:00.000Z', 'brand new task'), + asstLine('msg-z', '2026-05-01T12:00:02.000Z', { input_tokens: 500, output_tokens: 120 }, [readBlock('/z.ts')]), + ].join('\n') + '\n' + await writeFile(sessionPath, replaced) + expect((await stat(sessionPath)).ino).not.toBe(inoBefore) + + readLineCalls.length = 0 + const warm = await parseWith(warmCache) + // inode changed => modified => full re-parse from byte 0, never an offset. + expect(offsetsFor(sessionPath).every(o => o === undefined || o === 0)).toBe(true) + + const cold = await coldFullReparse() + expect(warm).toEqual(cold) + await rm(warmCache, { recursive: true, force: true }) + }) + + it('EDGE: cached offset beyond current EOF falls back to a full re-parse', async () => { + const warmCache = await mkdtemp(join(tmpdir(), 'incr-warm5-')) + await writeFile(sessionPath, baseLines().join('\n') + '\n') + await parseWith(warmCache) + + // Corrupt the persisted offset to point far beyond the file, then grow it. + const cachePath = sessionCachePath() + const cache = JSON.parse(await readFile(cachePath, 'utf-8')) + cache.providers.claude.files[sessionPath].lastCompleteLineOffset = 10_000_000 + await writeFile(cachePath, JSON.stringify(cache)) + + await appendFile(sessionPath, + userLine('2026-05-01T13:00:00.000Z', 'grow the file') + '\n' + + asstLine('msg-y', '2026-05-01T13:00:02.000Z', { input_tokens: 400, output_tokens: 100 }, [bashBlock('pwd')]) + '\n') + + readLineCalls.length = 0 + const warm = await parseWith(warmCache) + // guard: never resume from the stranded offset. + expect(offsetsFor(sessionPath).some(o => o === 10_000_000)).toBe(false) + + const cold = await coldFullReparse() + expect(warm).toEqual(cold) + await rm(warmCache, { recursive: true, force: true }) + }) +}) diff --git a/tests/session-cache.test.ts b/tests/session-cache.test.ts index e06fd7cb..2c736b22 100644 --- a/tests/session-cache.test.ts +++ b/tests/session-cache.test.ts @@ -387,6 +387,17 @@ describe('reconcileFile', () => { expect(reconcileFile(changed, marker)).toEqual({ action: 'modified' }) }) + it('returns "modified" when the cached offset is stranded beyond the current EOF', () => { + // A truncate-then-regrow can leave the resume offset past live bytes; resuming + // there would drop the appended tail, so it must fall back to a full re-parse. + const cached = makeCachedFile({ + fingerprint: { dev: 1, ino: 100, mtimeMs: 1000, sizeBytes: 5000 }, + lastCompleteLineOffset: 9_000_000, + }) + const current: FileFingerprint = { dev: 1, ino: 100, mtimeMs: 2000, sizeBytes: 8000 } + expect(reconcileFile(current, cached)).toEqual({ action: 'modified' }) + }) + it('returns "modified" when size shrank', () => { const cached = makeCachedFile({ fingerprint: { dev: 1, ino: 100, mtimeMs: 1000, sizeBytes: 5000 },