codeburn/packages/cli/tests/parser-incremental-append.test.ts

374 lines
18 KiB
TypeScript

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 } })
function prLinkLine(ts: string, url: string): string {
return JSON.stringify({ type: 'pr-link', sessionId: 'sess-1', timestamp: ts, cwd: CWD, prUrl: url })
}
// 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('PR-REFS: survive the incremental append path (continuation merge unions refs)', async () => {
const warmCache = await mkdtemp(join(tmpdir(), 'incr-pr-'))
// Base: one turn that creates PR-1.
await writeFile(sessionPath,
userLine('2026-05-01T10:00:01.000Z', 'ship PR one') + '\n' +
asstLine('msg-a', '2026-05-01T10:00:02.000Z', { input_tokens: 100, output_tokens: 20 }, [bashBlock('gh pr create')]) + '\n' +
prLinkLine('2026-05-01T10:00:03.000Z', 'https://github.com/o/r/pull/1') + '\n')
await parseWith(warmCache)
// Append a continuation of that same turn (no leading user message) that
// references PR-2, then a fresh turn that references PR-3.
await appendFile(sessionPath,
asstLine('msg-b', '2026-05-01T10:00:04.000Z', { input_tokens: 50, output_tokens: 10 }, [bashBlock('gh pr create')]) + '\n' +
prLinkLine('2026-05-01T10:00:05.000Z', 'https://github.com/o/r/pull/2') + '\n' +
userLine('2026-05-01T10:10:00.000Z', 'ship PR three') + '\n' +
asstLine('msg-c', '2026-05-01T10:10:02.000Z', { input_tokens: 80, output_tokens: 20 }, [bashBlock('gh pr create')]) + '\n' +
prLinkLine('2026-05-01T10:10:03.000Z', 'https://github.com/o/r/pull/3') + '\n')
readLineCalls.length = 0
const warm = await parseWith(warmCache)
expect(offsetsFor(sessionPath).some(o => o !== undefined && o > 0)).toBe(true) // took the append path
const cold = await coldFullReparse()
expect(warm).toEqual(cold)
const turns = warm[0]!.sessions[0]!.turns
expect(turns[0]!.prRefs).toEqual(['https://github.com/o/r/pull/1', 'https://github.com/o/r/pull/2'])
expect(turns[1]!.prRefs).toEqual(['https://github.com/o/r/pull/3'])
await rm(warmCache, { recursive: true, force: true })
})
it('PR-REFS: survive when a straddled append falls back to a full re-parse', async () => {
const warmCache = await mkdtemp(join(tmpdir(), 'incr-pr2-'))
await writeFile(sessionPath,
userLine('2026-05-01T10:00:01.000Z', 'ship PR one') + '\n' +
asstLine('msg-a', '2026-05-01T10:00:02.000Z', { input_tokens: 100, output_tokens: 20 }, [bashBlock('gh pr create')]) + '\n' +
prLinkLine('2026-05-01T10:00:03.000Z', 'https://github.com/o/r/pull/1') + '\n')
await parseWith(warmCache)
// Re-emit msg-a (an id already committed in the cached prefix) -> straddle ->
// the shortcut is abandoned and the file re-parses from byte 0.
await appendFile(sessionPath,
asstLine('msg-a', '2026-05-01T10:00:02.500Z', { input_tokens: 100, output_tokens: 40 }, [bashBlock('gh pr create')]) + '\n' +
prLinkLine('2026-05-01T10:00:06.000Z', 'https://github.com/o/r/pull/2') + '\n')
const warm = await parseWith(warmCache)
const cold = await coldFullReparse()
expect(warm).toEqual(cold)
expect(warm[0]!.sessions[0]!.turns[0]!.prRefs).toEqual([
'https://github.com/o/r/pull/1', 'https://github.com/o/r/pull/2',
])
await rm(warmCache, { recursive: true, force: true })
})
it('SUBAGENT-LINKS: spawnToolUseIds + agentSpawnLinks survive the incremental append path', async () => {
const warmCache = await mkdtemp(join(tmpdir(), 'incr-spawn-'))
const agentBlock = (id: string) => ({ type: 'tool_use', name: 'Agent', id, input: {} })
const spawnResultLine = (ts: string, toolUseId: string, agentId: string): string =>
JSON.stringify({
type: 'user', sessionId: 'sess-1', timestamp: ts, cwd: CWD,
message: { role: 'user', content: [{ type: 'tool_result', tool_use_id: toolUseId, content: 'agent done' }] },
toolUseResult: { status: 'completed', agentId },
})
// Base: one turn spawns agent1; its result records the agentId->tool_use link.
await writeFile(sessionPath,
userLine('2026-05-01T10:00:01.000Z', 'launch a reviewer') + '\n' +
asstLine('msg-a', '2026-05-01T10:00:02.000Z', { input_tokens: 100, output_tokens: 20 }, [agentBlock('toolu_a1')]) + '\n' +
spawnResultLine('2026-05-01T10:00:03.000Z', 'toolu_a1', 'agent1') + '\n')
await parseWith(warmCache)
// Append: a continuation of that same turn spawns agent2 (merged into turn 0),
// then a fresh turn spawns agent3.
await appendFile(sessionPath,
asstLine('msg-b', '2026-05-01T10:00:04.000Z', { input_tokens: 50, output_tokens: 10 }, [agentBlock('toolu_a2')]) + '\n' +
spawnResultLine('2026-05-01T10:00:05.000Z', 'toolu_a2', 'agent2') + '\n' +
userLine('2026-05-01T10:10:00.000Z', 'launch another') + '\n' +
asstLine('msg-c', '2026-05-01T10:10:02.000Z', { input_tokens: 80, output_tokens: 20 }, [agentBlock('toolu_a3')]) + '\n' +
spawnResultLine('2026-05-01T10:10:03.000Z', 'toolu_a3', 'agent3') + '\n')
readLineCalls.length = 0
const warm = await parseWith(warmCache)
expect(offsetsFor(sessionPath).some(o => o !== undefined && o > 0)).toBe(true) // took the append path
const cold = await coldFullReparse()
expect(warm).toEqual(cold)
const session = warm[0]!.sessions[0]!
expect(session.turns[0]!.spawnToolUseIds).toEqual(['toolu_a1', 'toolu_a2'])
expect(session.turns[1]!.spawnToolUseIds).toEqual(['toolu_a3'])
expect(session.agentSpawnLinks).toEqual({ agent1: 'toolu_a1', agent2: 'toolu_a2', agent3: 'toolu_a3' })
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 })
})
})
describe('straddle guard (streamed id restated across the append boundary)', () => {
// A streamed assistant id whose first emission sits in the committed prefix
// (and NOT in the last cached turn) can be restated in the appended region.
// The boundary merge would splice it into the wrong turn and count it twice;
// the guard must abandon the shortcut and re-parse the file from byte 0,
// matching the cold-parse oracle exactly.
it('falls back to a full re-parse when an appended id already exists in the cached turns', async () => {
const cacheDir = await mkdtemp(join(tmpdir(), 'incr-straddle-'))
try {
await writeFile(sessionPath, baseLines().join('\n') + '\n')
await parseWith(cacheDir)
// Restatement of msg-a (first turn's id, grown usage) followed by a new
// turn — the shape image-heavy sessions produce while streaming.
const appended = [
asstLine('msg-a', '2026-05-01T10:06:00.000Z', { input_tokens: 100, output_tokens: 90, cache_read_input_tokens: 300 }, [readBlock('/a.ts')]),
userLine('2026-05-01T10:07:00.000Z', 'third task please'),
asstLine('msg-c', '2026-05-01T10:07:02.000Z', { input_tokens: 50, output_tokens: 10 }, []),
]
await appendFile(sessionPath, appended.join('\n') + '\n')
readLineCalls.length = 0
const incremental = await parseWith(cacheDir)
const oracle = await coldFullReparse()
const sum = (ps: ProjectSummary[]) => ({
calls: ps.reduce((s, p) => s + p.totalApiCalls, 0),
cost: ps.reduce((s, p) => s + p.totalCostUSD, 0),
turns: ps.flatMap(p => p.sessions).reduce((s, x) => s + x.turns.length, 0),
})
expect(sum(incremental)).toEqual(sum(oracle))
// The guard must have re-read the file from byte 0, not the append offset.
const offsets = offsetsFor(sessionPath)
expect(offsets.some(o => o === undefined || o === 0)).toBe(true)
} finally {
await rm(cacheDir, { recursive: true, force: true })
}
})
})