From 50dca74fc284e0ed1680ba4b7f95a7fedcf56537 Mon Sep 17 00:00:00 2001 From: kimlj Date: Fri, 23 Oct 2026 12:35:08 +0800 Subject: [PATCH] Keep every parallel tool call of a streamed Claude message Claude Code writes one JSONL line per content block, so a message that calls several tools in parallel spans several lines with the same message id. dedupeStreamingMessageIds kept only the last line, which holds the right usage but only the last tool_use. The earlier tool calls were dropped from the tool, MCP and shell breakdowns and from turn classification. Keep the last line, and put back the tool_use blocks of the earlier lines, matched by block id so a stream that restates every block in its final write (#110) is not counted twice. Calls, tokens and cost are unchanged. Bump the claude parse tag so cached sessions re-parse. The daily cache version is left alone: v77 is the last bump the zstd split can carry. --- src/parser.ts | 38 ++++++++++- src/session-cache.ts | 4 +- tests/parser-streamed-tool-blocks.test.ts | 81 +++++++++++++++++++++++ 3 files changed, 120 insertions(+), 3 deletions(-) create mode 100644 tests/parser-streamed-tool-blocks.test.ts 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) + }) +})