Skip to content

parser: full re-parse when an appended region restates a cached message id #772

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
iamtoruk merged 1 commit into from
Jul 20, 2026
Merged

parser: full re-parse when an appended region restates a cached message id #772

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
138 changes: 77 additions & 61 deletions src/parser.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1880,71 +1880,87 @@ async function scanProjectDirs(
const newEntries = await parseClaudeEntries(filePath, tracker, append.readFromOffset, { toolResultMeta, sessionMeta })
const cached = append.cached

const newTurns = newEntries
? parsedTurnsToCachedTurns(groupIntoTurns(dedupeStreamingMessageIds(newEntries), seenMsgIds, toolResultMeta))
: []

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
// Straddle guard: a streamed assistant message id that first appeared in
// the committed prefix can be restated inside the appended region
// (image-heavy turns stream one id across several records over seconds).
// The appended region is grouped before this file's cached keys join
// seenMsgIds, so the restated id would count twice; suppressing it
// instead would freeze the stale first emission. Neither matches a full
// re-parse, so on any id overlap the shortcut is abandoned and the file
// re-parses from byte 0 (rare: ~0.3% of real files).
const cachedIds = new Set(cached.turns.flatMap(t => t.calls.map(c => c.deduplicationKey)))
const straddles = newEntries !== null && newEntries.some(e => {
const id = getMessageId(e)
return id !== null && cachedIds.has(id)
})
if (!straddles) {
const newTurns = newEntries
? parsedTurnsToCachedTurns(groupIntoTurns(dedupeStreamingMessageIds(newEntries), seenMsgIds, toolResultMeta))
: []

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]!)
}
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
}
// 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

// Session meta merges across the append boundary: title is last-wins
// (prefer the newly-parsed tail), PR links union, isSidechain is sticky.
const mergedTitle = sessionMeta.title ?? cached.title
const mergedPrLinks = Array.from(new Set([...(cached.prLinks ?? []), ...sessionMeta.prLinks]))
const mergedSidechain = cached.isSidechain === true || sessionMeta.isSidechain

section.files[filePath] = {
fingerprint: info.fp,
lastCompleteLineOffset: tracker.lastCompleteLineOffset,
canonicalCwd,
canonicalProjectName,
mcpInventory,
turns: mergedTurns,
agentType: cached.agentType,
...(mergedTitle ? { title: mergedTitle } : {}),
...(mergedPrLinks.length > 0 ? { prLinks: mergedPrLinks } : {}),
...(mergedSidechain ? { isSidechain: true } : {}),
}
;(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 })
// 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

// Session meta merges across the append boundary: title is last-wins
// (prefer the newly-parsed tail), PR links union, isSidechain is sticky.
const mergedTitle = sessionMeta.title ?? cached.title
const mergedPrLinks = Array.from(new Set([...(cached.prLinks ?? []), ...sessionMeta.prLinks]))
const mergedSidechain = cached.isSidechain === true || sessionMeta.isSidechain

section.files[filePath] = {
fingerprint: info.fp,
lastCompleteLineOffset: tracker.lastCompleteLineOffset,
canonicalCwd,
canonicalProjectName,
mcpInventory,
turns: mergedTurns,
agentType: cached.agentType,
...(mergedTitle ? { title: mergedTitle } : {}),
...(mergedPrLinks.length > 0 ? { prLinks: mergedPrLinks } : {}),
...(mergedSidechain ? { isSidechain: true } : {}),
}
;(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
}
if (onFileParsed) await onFileParsed()
continue
// Straddled: fall through to the full re-parse below.
}

const tracker = { lastCompleteLineOffset: 0 }
Expand Down
41 changes: 41 additions & 0 deletions tests/parser-incremental-append.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,3 +236,44 @@ describe('incremental append parsing', () => {
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 })
}
})
})