diff --git a/src/parser.ts b/src/parser.ts index 26154921c..9e250c1ba 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -1658,11 +1658,15 @@ export function parseAdvisorCalls(entry: JournalEntry): ParsedApiCall[] { export function dedupeStreamingMessageIds(entries: JournalEntry[]): JournalEntry[] { const firstIdxById = new Map() const lastIdxById = new Map() + const idxsById = new Map() 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[] = [] @@ -1670,8 +1674,10 @@ export function dedupeStreamingMessageIds(entries: JournalEntry[]): JournalEntry 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]!) @@ -1679,6 +1685,34 @@ export function dedupeStreamingMessageIds(entries: JournalEntry[]): JournalEntry 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() + 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, toolResultMeta?: Map): ParsedTurn[] { const turns: ParsedTurn[] = [] let currentUserMessage = '' diff --git a/src/session-cache.ts b/src/session-cache.ts index 08051906a..7698e432c 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -377,7 +377,9 @@ export const PROVIDER_PARSE_VERSIONS: Record = { // 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 diff --git a/tests/parser-streamed-tool-blocks.test.ts b/tests/parser-streamed-tool-blocks.test.ts new file mode 100644 index 000000000..5c51e8dda --- /dev/null +++ b/tests/parser-streamed-tool-blocks.test.ts @@ -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) + }) +})