Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
38 changes: 36 additions & 2 deletions src/parser.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1658,27 +1658,61 @@ export function parseAdvisorCalls(entry: JournalEntry): ParsedApiCall[] {
export function dedupeStreamingMessageIds(entries: JournalEntry[]): JournalEntry[] {
const firstIdxById = new Map<string, number>()
const lastIdxById = new Map<string, number>()
const idxsById = new Map<string, number[]>()
for (let i = 0; i < entries.length; i++) {
const id = getMessageId(entries[i]!)
if (!id) continue
if (!firstIdxById.has(id)) firstIdxById.set(id, i)
lastIdxById.set(id, i)
const idxs = idxsById.get(id)
if (idxs) idxs.push(i)
else idxsById.set(id, [i])
}
if (lastIdxById.size === 0) return entries
const result: JournalEntry[] = []
for (let i = 0; i < entries.length; i++) {
const id = getMessageId(entries[i]!)
if (id && lastIdxById.get(id) !== i) continue
if (id && firstIdxById.get(id) !== i) {
const firstTs = entries[firstIdxById.get(id)!]!.timestamp
result.push({ ...entries[i]!, timestamp: firstTs ?? entries[i]!.timestamp })
const firstIdx = firstIdxById.get(id)!
const firstTs = entries[firstIdx]!.timestamp
const message = withEarlierToolUses(entries, idxsById.get(id)!)
result.push({ ...entries[i]!, timestamp: firstTs ?? entries[i]!.timestamp, ...(message ? { message } : {}) })
continue
}
result.push(entries[i]!)
}
return result
}

// Claude Code writes one line per content block, so the parallel tool calls of
// one message sit on separate lines and the last line holds only the last of
// them. Keeping the last line alone keeps the authoritative usage but drops the
// earlier tool_use blocks, and with them their tool, MCP and Bash counts. This
// returns the last line's message with those earlier blocks put back in order,
// matched by block id so a stream that restates every block in its final write
// (#110) is not counted twice. Null when nothing was missing.
function withEarlierToolUses(entries: JournalEntry[], idxs: number[]): AssistantMessageContent | null {
const last = entries[idxs[idxs.length - 1]!]!.message as AssistantMessageContent
const lastContent = Array.isArray(last.content) ? last.content : []
const present = new Set<string>()
for (const block of lastContent) {
if (block.type === 'tool_use' && typeof block.id === 'string') present.add(block.id)
}
const missing: ContentBlock[] = []
for (const j of idxs.slice(0, -1)) {
const content = (entries[j]!.message as AssistantMessageContent).content
if (!Array.isArray(content)) continue
for (const block of content) {
if (block.type !== 'tool_use' || typeof block.id !== 'string' || present.has(block.id)) continue
present.add(block.id)
missing.push(block)
}
}
if (missing.length === 0) return null
return { ...last, content: [...missing, ...lastContent] }
}

export function groupIntoTurns(entries: JournalEntry[], seenMsgIds: Set<string>, toolResultMeta?: Map<string, ToolResultMeta>): ParsedTurn[] {
const turns: ParsedTurn[] = []
let currentUserMessage = ''
Expand Down
4 changes: 3 additions & 1 deletion src/session-cache.ts
Original file line number Diff line number Diff line change
Expand Up @@ -377,7 +377,9 @@ export const PROVIDER_PARSE_VERSIONS: Record<string, string> = {
// queued_command prompt attachments, including classification and PR links.
// per-call-cwd-v1: cached calls carry their cwd so a session that moved
// folders splits across projects per call.
claude: 'advisor-usage-v1-skills-rich-capture-v1-cross-provider-pr-v1-session-lineage-capture-v1-queued-human-prompts-v1-per-call-cwd-v1',
// streamed-tool-blocks-v1: parallel tool_use blocks written on earlier lines
// of one streamed message are kept, so cached calls regain their tools.
claude: 'advisor-usage-v1-skills-rich-capture-v1-cross-provider-pr-v1-session-lineage-capture-v1-queued-human-prompts-v1-per-call-cwd-v1-streamed-tool-blocks-v1',
cline: 'worktree-project-grouping-v1',
// reported-cost-v1: the CLI reports its own per-message cost, so entries
// cached before cline-cli joined the reported-cost allowlist in parser.ts
Expand Down
81 changes: 81 additions & 0 deletions tests/parser-streamed-tool-blocks.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
import { describe, expect, it } from 'vitest'

import { dedupeStreamingMessageIds, parseApiCall } from '../src/parser.js'
import type { ContentBlock, JournalEntry } from '../src/types.js'

// Claude Code writes one JSONL line per content block of an assistant message:
// every line repeats the message id and usage, and each carries only its own
// block. Parallel tool calls therefore arrive as several lines with the same id.
// Older streams (#110) instead restate the message and put every block in the
// final write. Both shapes must yield each tool_use exactly once.

const usage = { input_tokens: 10, output_tokens: 200, cache_read_input_tokens: 5_000, cache_creation_input_tokens: 0 }

function line(id: string, content: ContentBlock[], timestamp: string): JournalEntry {
return {
type: 'assistant',
sessionId: 's1',
timestamp,
message: { id, type: 'message', role: 'assistant', model: 'claude-sonnet-4-5', content, usage },
}
}

const read = (id: string): ContentBlock => ({ type: 'tool_use', id, name: 'Read', input: { file_path: `/work/${id}.ts` } })
const bash = (id: string): ContentBlock => ({ type: 'tool_use', id, name: 'Bash', input: { command: 'npm test' } })
const mcp = (id: string): ContentBlock => ({ type: 'tool_use', id, name: 'mcp__playwright__browser_navigate', input: { url: 'http://localhost' } })

function toolsOf(entries: JournalEntry[]): string[][] {
return dedupeStreamingMessageIds(entries).map(e => parseApiCall(e)?.tools ?? [])
}

describe('streamed assistant messages with one content block per line', () => {
it('keeps every parallel tool_use, not only the last line', () => {
const entries = [
line('msg-1', [{ type: 'thinking', thinking: 'plan' }], '2026-09-11T10:00:00.000Z'),
line('msg-1', [read('t1')], '2026-09-11T10:00:01.000Z'),
line('msg-1', [read('t2')], '2026-09-11T10:00:02.000Z'),
line('msg-1', [bash('t3')], '2026-09-11T10:00:03.000Z'),
]
const deduped = dedupeStreamingMessageIds(entries)
expect(deduped).toHaveLength(1)
expect(deduped[0]!.timestamp).toBe('2026-09-11T10:00:00.000Z')
expect(toolsOf(entries)).toEqual([['Read', 'Read', 'Bash']])
})

it('keeps an MCP call made in parallel with a built-in tool', () => {
const entries = [
line('msg-1', [mcp('t1')], '2026-09-11T10:00:01.000Z'),
line('msg-1', [read('t2')], '2026-09-11T10:00:02.000Z'),
]
const call = parseApiCall(dedupeStreamingMessageIds(entries)[0]!)
expect(call?.mcpTools).toEqual(['mcp__playwright__browser_navigate'])
expect(call?.tools).toContain('Read')
})

it('still counts usage once per message', () => {
const entries = [
line('msg-1', [read('t1')], '2026-09-11T10:00:01.000Z'),
line('msg-1', [bash('t2')], '2026-09-11T10:00:02.000Z'),
]
const calls = dedupeStreamingMessageIds(entries).map(e => parseApiCall(e)!)
expect(calls).toHaveLength(1)
expect(calls[0]!.usage.outputTokens).toBe(200)
})
})

describe('streamed assistant messages restated with every block in the final write (#110)', () => {
it('does not double count a tool_use repeated in a later restatement', () => {
const entries = [
line('msg-1', [], '2026-09-11T10:00:00.000Z'),
line('msg-1', [read('t1')], '2026-09-11T10:00:01.000Z'),
line('msg-1', [read('t1'), bash('t2')], '2026-09-11T10:00:02.000Z'),
]
expect(toolsOf(entries)).toEqual([['Read', 'Bash']])
})

it('leaves a message whose last line already holds every block unchanged', () => {
const last = line('msg-1', [read('t1'), bash('t2')], '2026-09-11T10:00:02.000Z')
const entries = [line('msg-1', [], '2026-09-11T10:00:00.000Z'), last]
expect(dedupeStreamingMessageIds(entries)[0]!.message).toBe(last.message)
})
})
Loading