Merge pull request #749 from getagentseal/perf/incremental-append-parse

perf(parser): incremental parse of append-grown session files

src/parser.ts

@@ -1593,7 +1593,7 @@ async function scanProjectDirs(
 
   type FileInfo = { dirName: string; fp: NonNullable<Awaited<ReturnType<typeof fingerprintFile>>>; 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:<ts>` 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<string, number>()
+  const lastIdx = new Map<string, number>()
+  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<JournalEntry[] | null> {
   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)

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

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<typeof import('../src/fs-utils.js')>()
+  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<string, number>,
+  blocks: Array<Record<string, unknown>> = [],
+  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<ProjectSummary[]> {
+  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<ProjectSummary[]> {
+  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<number | undefined> {
+  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 })
+  })
+})

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 },