From d6e6db0387a2274afd847196fcc99846935770dc Mon Sep 17 00:00:00 2001 From: fullsend-code <278716306+fullsend-ai-coder[bot]@users.noreply.github.com> Date: Wed, 2 Sep 2026 10:47:11 +0000 Subject: [PATCH 1/3] fix(#6904): parse thinking tokens from Claude stream events The Claude runtime parser (claude_progress.go) never populated ReasoningTokens because the message_delta handler only parsed usage.output_tokens, ignoring output_tokens_details.thinking_tokens from the Anthropic API. Add parsing of output_tokens_details.thinking_tokens from message_delta events and accumulate across turns. The per-message value is emitted on TokensEvent for progress display, and the accumulated total is emitted on ResultEvent so gen_ai.usage. reasoning_tokens appears on OTel spans for Claude Code runs. This closes the observability gap where Pi runs reported reasoning tokens but Claude Code runs always showed 0. Closes #6904 --- internal/runtime/claude_progress.go | 20 +++-- internal/runtime/claude_progress_test.go | 105 +++++++++++++++++++++++ 2 files changed, 120 insertions(+), 5 deletions(-) diff --git a/internal/runtime/claude_progress.go b/internal/runtime/claude_progress.go index 72142d01bf..414d2ba3ee 100644 --- a/internal/runtime/claude_progress.go +++ b/internal/runtime/claude_progress.go @@ -118,6 +118,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { // per-message token tracking for throttled TokensEvent totalInput int totalOutput int + totalReasoning int // per-message thinking tokens (reset on message_start) totalCacheRead int totalCacheWrite int lastEmittedTotal int @@ -128,6 +129,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { cumulativeCacheRead int cumulativeCacheWrite int seenResult bool + accReasoning int // accumulated thinking tokens across all messages (for ResultEvent) ) // Emit a final cumulative TokensEvent when the stream ends without @@ -275,6 +277,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { totalInput = msg.Message.Usage.InputTokens totalOutput = 0 + totalReasoning = 0 totalCacheRead = msg.Message.Usage.CacheReadInputTokens totalCacheWrite = msg.Message.Usage.CacheCreationInputTokens } @@ -282,20 +285,26 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { case "message_delta": var md struct { Usage struct { - OutputTokens int `json:"output_tokens"` + OutputTokens int `json:"output_tokens"` + OutputTokensDetails struct { + ThinkingTokens int `json:"thinking_tokens"` + } `json:"output_tokens_details"` } `json:"usage"` } if err := json.Unmarshal(wrapper.Event, &md); err == nil && md.Usage.OutputTokens > 0 { totalOutput = md.Usage.OutputTokens + totalReasoning = md.Usage.OutputTokensDetails.ThinkingTokens + accReasoning += totalReasoning total := cumulativeInput + totalInput + cumulativeOutput + totalOutput + cumulativeCacheRead + totalCacheRead + cumulativeCacheWrite + totalCacheWrite if total-lastEmittedTotal >= tokenThreshold { lastEmittedTotal = total onEvent(TokensEvent{ - InputTokens: cumulativeInput + totalInput, - OutputTokens: cumulativeOutput + totalOutput, - CacheRead: cumulativeCacheRead + totalCacheRead, - CacheWrite: cumulativeCacheWrite + totalCacheWrite, + InputTokens: cumulativeInput + totalInput, + OutputTokens: cumulativeOutput + totalOutput, + ReasoningTokens: totalReasoning, + CacheRead: cumulativeCacheRead + totalCacheRead, + CacheWrite: cumulativeCacheWrite + totalCacheWrite, }) } } @@ -315,6 +324,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { Subtype: re.Subtype, InputTokens: re.Usage.InputTokens, OutputTokens: re.Usage.OutputTokens, + ReasoningTokens: accReasoning, CacheCreationInputTokens: re.Usage.CacheCreationInputTokens, CacheReadInputTokens: re.Usage.CacheReadInputTokens, }) diff --git a/internal/runtime/claude_progress_test.go b/internal/runtime/claude_progress_test.go index 3aed0969ae..33c4a21b58 100644 --- a/internal/runtime/claude_progress_test.go +++ b/internal/runtime/claude_progress_test.go @@ -942,6 +942,111 @@ func TestParseClaudeStreamTokensEvent(t *testing.T) { } } +func TestParseClaudeStreamTokensEventWithReasoningTokens(t *testing.T) { + lines := []string{ + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":4000,"cache_read_input_tokens":500,"cache_creation_input_tokens":200}}}}`, + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":1000,"output_tokens_details":{"thinking_tokens":300}}}}`, + } + events := collectEvents(t, strings.Join(lines, "\n")) + + var tokens []TokensEvent + for _, e := range events { + if te, ok := e.(TokensEvent); ok { + tokens = append(tokens, te) + } + } + // Total = 4000 + 1000 + 500 + 200 = 5700, crosses 5k threshold + if len(tokens) != 1 { + t.Fatalf("expected 1 tokens event, got %d", len(tokens)) + } + if tokens[0].ReasoningTokens != 300 { + t.Errorf("expected 300 reasoning tokens, got %d", tokens[0].ReasoningTokens) + } + if tokens[0].OutputTokens != 1000 { + t.Errorf("expected 1000 output tokens, got %d", tokens[0].OutputTokens) + } +} + +func TestParseClaudeStreamResultEventAccumulatesReasoningTokens(t *testing.T) { + lines := []string{ + // First message turn with 200 thinking tokens. + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":4000,"cache_read_input_tokens":500,"cache_creation_input_tokens":200}}}}`, + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":1000,"output_tokens_details":{"thinking_tokens":200}}}}`, + // Second message turn with 150 thinking tokens. + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":6000,"cache_read_input_tokens":500,"cache_creation_input_tokens":200}}}}`, + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":800,"output_tokens_details":{"thinking_tokens":150}}}}`, + // Result event. + `{"type":"result","num_turns":2,"total_cost_usd":0.50,"usage":{"input_tokens":10000,"output_tokens":1800,"cache_creation_input_tokens":400,"cache_read_input_tokens":1000}}`, + } + events := collectEvents(t, strings.Join(lines, "\n")) + + var results []ResultEvent + for _, e := range events { + if re, ok := e.(ResultEvent); ok { + results = append(results, re) + } + } + if len(results) != 1 { + t.Fatalf("expected 1 result event, got %d", len(results)) + } + // Accumulated: 200 + 150 = 350. + if results[0].ReasoningTokens != 350 { + t.Errorf("expected 350 accumulated reasoning tokens, got %d", results[0].ReasoningTokens) + } +} + +func TestParseClaudeStreamNoThinkingTokensBackwardCompat(t *testing.T) { + lines := []string{ + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":4000,"cache_read_input_tokens":500,"cache_creation_input_tokens":200}}}}`, + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":1000}}}`, + `{"type":"result","num_turns":1,"total_cost_usd":0.10,"usage":{"input_tokens":4000,"output_tokens":1000,"cache_creation_input_tokens":200,"cache_read_input_tokens":500}}`, + } + events := collectEvents(t, strings.Join(lines, "\n")) + + var tokens []TokensEvent + var results []ResultEvent + for _, e := range events { + switch ev := e.(type) { + case TokensEvent: + tokens = append(tokens, ev) + case ResultEvent: + results = append(results, ev) + } + } + // TokensEvent: reasoning should be 0 when no thinking tokens present. + if len(tokens) == 1 && tokens[0].ReasoningTokens != 0 { + t.Errorf("expected 0 reasoning tokens when absent, got %d", tokens[0].ReasoningTokens) + } + // ResultEvent: reasoning should be 0. + if len(results) != 1 { + t.Fatalf("expected 1 result event, got %d", len(results)) + } + if results[0].ReasoningTokens != 0 { + t.Errorf("expected 0 reasoning tokens in result when absent, got %d", results[0].ReasoningTokens) + } +} + +func TestProgressParserCapturesReasoningTokensInMetrics(t *testing.T) { + lines := []string{ + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":4000,"cache_read_input_tokens":500,"cache_creation_input_tokens":200}}}}`, + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":1000,"output_tokens_details":{"thinking_tokens":250}}}}`, + `{"type":"result","num_turns":1,"total_cost_usd":0.10,"usage":{"input_tokens":4000,"output_tokens":1000,"cache_creation_input_tokens":200,"cache_read_input_tokens":500}}`, + } + + input := strings.NewReader(strings.Join(lines, "\n")) + var buf bytes.Buffer + printer := ui.New(&buf) + metrics := &RunMetrics{} + + if err := progressParser(input, printer, metrics); err != nil { + t.Fatalf("progressParser returned error: %v", err) + } + + if metrics.ReasoningTokens != 250 { + t.Errorf("expected 250 reasoning tokens in metrics, got %d", metrics.ReasoningTokens) + } +} + func TestParseClaudeStreamTokensEventThrottled(t *testing.T) { lines := []string{ `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":4000}}}}`, From 9caee858ab2bb3bfa87162f74d0605f1d424f1e8 Mon Sep 17 00:00:00 2001 From: fullsend-fix <278716306+fullsend-ai-coder[bot]@users.noreply.github.com> Date: Wed, 2 Sep 2026 11:15:56 +0000 Subject: [PATCH 2/3] fix(runtime): align reasoning token naming and throttle calculation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rename totalReasoning → msgReasoning (per-message, reset on message_start) and accReasoning → totalReasoning (cross-message accumulator) to match the naming convention in pi_progress.go and opencode_progress.go. Include msgReasoning in the throttle threshold so TokensEvent emissions are consistent with the renderer's total. Addresses #6907 --- internal/runtime/claude_progress.go | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/internal/runtime/claude_progress.go b/internal/runtime/claude_progress.go index 414d2ba3ee..c0192d59c1 100644 --- a/internal/runtime/claude_progress.go +++ b/internal/runtime/claude_progress.go @@ -118,7 +118,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { // per-message token tracking for throttled TokensEvent totalInput int totalOutput int - totalReasoning int // per-message thinking tokens (reset on message_start) + msgReasoning int // per-message thinking tokens (reset on message_start) totalCacheRead int totalCacheWrite int lastEmittedTotal int @@ -129,7 +129,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { cumulativeCacheRead int cumulativeCacheWrite int seenResult bool - accReasoning int // accumulated thinking tokens across all messages (for ResultEvent) + totalReasoning int // accumulated thinking tokens across all messages (for ResultEvent) ) // Emit a final cumulative TokensEvent when the stream ends without @@ -277,7 +277,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { totalInput = msg.Message.Usage.InputTokens totalOutput = 0 - totalReasoning = 0 + msgReasoning = 0 totalCacheRead = msg.Message.Usage.CacheReadInputTokens totalCacheWrite = msg.Message.Usage.CacheCreationInputTokens } @@ -293,16 +293,16 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { } if err := json.Unmarshal(wrapper.Event, &md); err == nil && md.Usage.OutputTokens > 0 { totalOutput = md.Usage.OutputTokens - totalReasoning = md.Usage.OutputTokensDetails.ThinkingTokens - accReasoning += totalReasoning + msgReasoning = md.Usage.OutputTokensDetails.ThinkingTokens + totalReasoning += msgReasoning total := cumulativeInput + totalInput + cumulativeOutput + totalOutput + - cumulativeCacheRead + totalCacheRead + cumulativeCacheWrite + totalCacheWrite + msgReasoning + cumulativeCacheRead + totalCacheRead + cumulativeCacheWrite + totalCacheWrite if total-lastEmittedTotal >= tokenThreshold { lastEmittedTotal = total onEvent(TokensEvent{ InputTokens: cumulativeInput + totalInput, OutputTokens: cumulativeOutput + totalOutput, - ReasoningTokens: totalReasoning, + ReasoningTokens: msgReasoning, CacheRead: cumulativeCacheRead + totalCacheRead, CacheWrite: cumulativeCacheWrite + totalCacheWrite, }) @@ -324,7 +324,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { Subtype: re.Subtype, InputTokens: re.Usage.InputTokens, OutputTokens: re.Usage.OutputTokens, - ReasoningTokens: accReasoning, + ReasoningTokens: totalReasoning, CacheCreationInputTokens: re.Usage.CacheCreationInputTokens, CacheReadInputTokens: re.Usage.CacheReadInputTokens, }) From c3b88f4374391b74e1b28b7fafb135bb11089ea6 Mon Sep 17 00:00:00 2001 From: fullsend-fix <278716306+fullsend-ai-coder[bot]@users.noreply.github.com> Date: Wed, 2 Sep 2026 11:41:20 +0000 Subject: [PATCH 3/3] fix(runtime): correct threshold comment to include reasoning tokens The comment in TestParseClaudeStreamTokensEventWithReasoningTokens incorrectly stated the total as 5700 (omitting the 300 reasoning tokens from the sum). The actual threshold calculation includes msgReasoning: 4000 + 1000 + 300 + 500 + 200 = 6000. Addresses #6907 --- internal/runtime/claude_progress_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/runtime/claude_progress_test.go b/internal/runtime/claude_progress_test.go index 33c4a21b58..0bd7b1a7e6 100644 --- a/internal/runtime/claude_progress_test.go +++ b/internal/runtime/claude_progress_test.go @@ -955,7 +955,7 @@ func TestParseClaudeStreamTokensEventWithReasoningTokens(t *testing.T) { tokens = append(tokens, te) } } - // Total = 4000 + 1000 + 500 + 200 = 5700, crosses 5k threshold + // Total = 4000 + 1000 + 300 + 500 + 200 = 6000, crosses 5k threshold if len(tokens) != 1 { t.Fatalf("expected 1 tokens event, got %d", len(tokens)) }