From 8a418287876f56db410d5a93ff2df822c5853931 Mon Sep 17 00:00:00 2001 From: fullsend-code <278716306+fullsend-ai-coder[bot]@users.noreply.github.com> Date: Mon, 31 Aug 2026 16:01:37 +0000 Subject: [PATCH 1/2] fix(#6806): emit fallback ResultEvent when stream ends without result line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the Claude Code agent process terminates abnormally (e.g., killed by a signal after writing output but before emitting its final result NDJSON line), parseClaudeStream now emits a synthetic ResultEvent with the best-available metrics accumulated from message_start and message_delta events during the stream. Previously, metrics relied entirely on the "result" event emitted at the very end of the Claude Code NDJSON stream. If the process was killed before that line was written, RunMetrics stayed at zero values (num_turns: 0, all token counts: 0), producing a metrics.json with zeroed data despite the agent having done substantial work. The fix tracks per-turn token counts incrementally: each message_start accumulates input and cache tokens, and each subsequent message_start folds in the previous turn's output tokens. At EOF or read error, if no real result event was seen and at least one turn was observed, a fallback ResultEvent is emitted with subtype "stream_incomplete" and IsError: true. When a real result event IS present, the fallback is suppressed and the definitive values are used as before. TotalCostUSD remains zero in the fallback because it cannot be derived from token counts alone without pricing data. This is an acceptable limitation — non-zero token counts and num_turns are the primary improvement over the previous all-zeros behavior. Closes #6806 --- internal/runtime/claude_progress.go | 46 ++++++ internal/runtime/claude_progress_test.go | 185 +++++++++++++++++++++++ 2 files changed, 231 insertions(+) diff --git a/internal/runtime/claude_progress.go b/internal/runtime/claude_progress.go index bfab70e6ff..e0aebf3246 100644 --- a/internal/runtime/claude_progress.go +++ b/internal/runtime/claude_progress.go @@ -108,6 +108,13 @@ type resultEvent struct { // emits normalized AgentEvent values via the onEvent callback. It processes // system events, stream_event deltas (thinking, text, tool input JSON), // result events, errors, and assistant message fallback. +// +// When the stream ends (EOF or read error) without a "result" event and at +// least one conversation turn was observed, a fallback ResultEvent is emitted +// with the best-available token counts accumulated from message_start and +// message_delta events. This ensures metrics are non-zero when the agent +// process terminates abnormally after doing work but before emitting its +// final result line. See #6806. func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { br := bufio.NewReaderSize(r, streamBufSize) @@ -121,14 +128,43 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { totalCacheRead int totalCacheWrite int lastEmittedTotal int + // Cumulative tracking for fallback ResultEvent when the stream + // ends without a "result" line (agent crash / signal kill). + numTurns int + sawResult bool + cumulInput int + cumulOutput int + cumulCacheRead int + cumulCacheWrite int ) + // emitFallbackResult sends a synthetic ResultEvent from accumulated + // per-turn data when the real result event is missing. The last + // turn's output tokens must be folded in before calling. + emitFallbackResult := func() { + if sawResult || numTurns == 0 { + return + } + cumulOutput += totalOutput + onEvent(ResultEvent{ + NumTurns: numTurns, + IsError: true, + Subtype: "stream_incomplete", + InputTokens: cumulInput, + OutputTokens: cumulOutput, + CacheCreationInputTokens: cumulCacheWrite, + CacheReadInputTokens: cumulCacheRead, + }) + } + for { line, isPrefix, err := br.ReadLine() if err == io.EOF { + emitFallbackResult() return nil } if err != nil { + emitFallbackResult() return err } if isPrefix { @@ -236,10 +272,19 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { } `json:"message"` } if err := json.Unmarshal(wrapper.Event, &msg); err == nil { + // Accumulate previous turn's output before resetting. + cumulOutput += totalOutput + totalInput = msg.Message.Usage.InputTokens totalOutput = 0 totalCacheRead = msg.Message.Usage.CacheReadInputTokens totalCacheWrite = msg.Message.Usage.CacheCreationInputTokens + + // Accumulate this turn's input and cache tokens. + cumulInput += totalInput + cumulCacheRead += totalCacheRead + cumulCacheWrite += totalCacheWrite + numTurns++ } case "message_delta": @@ -264,6 +309,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { } case "result": + sawResult = true var re resultEvent if err := json.Unmarshal(line, &re); err != nil { continue diff --git a/internal/runtime/claude_progress_test.go b/internal/runtime/claude_progress_test.go index ce2521b46b..795049825c 100644 --- a/internal/runtime/claude_progress_test.go +++ b/internal/runtime/claude_progress_test.go @@ -633,6 +633,191 @@ func TestProgressParserNoResultEvent(t *testing.T) { } } +// TestParseClaudeStreamFallbackResultOnIncompleteStream verifies that when +// the stream ends without a "result" event but message_start events were +// observed, a fallback ResultEvent is emitted with accumulated token counts. +// This covers the scenario in #6806 where the agent process terminates +// abnormally after doing work but before emitting its final result line. +func TestParseClaudeStreamFallbackResultOnIncompleteStream(t *testing.T) { + lines := []string{ + // System init + `{"type":"system","subtype":"init","model":"claude-opus-4-6","claude_code_version":"1.0.50"}`, + // Turn 1: message_start with token usage + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":5000,"cache_read_input_tokens":3000,"cache_creation_input_tokens":1000}}}}`, + // Turn 1: content + `{"type":"stream_event","event":{"type":"content_block_start","content_block":{"type":"tool_use","name":"Read"}}}`, + `{"type":"stream_event","event":{"type":"content_block_stop"}}`, + // Turn 1: message_delta with output tokens + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":2000}}}`, + // Turn 2: message_start — new turn accumulates previous output + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":8000,"cache_read_input_tokens":4000,"cache_creation_input_tokens":0}}}}`, + // Turn 2: content + `{"type":"stream_event","event":{"type":"content_block_start","content_block":{"type":"tool_use","name":"Write"}}}`, + `{"type":"stream_event","event":{"type":"content_block_stop"}}`, + // Turn 2: output tokens + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":3000}}}`, + // No result event — stream ends (process killed) + } + + events := collectEvents(t, strings.Join(lines, "\n")) + + // Find the fallback ResultEvent. + var results []ResultEvent + for _, e := range events { + if r, ok := e.(ResultEvent); ok { + results = append(results, r) + } + } + if len(results) != 1 { + t.Fatalf("expected 1 fallback ResultEvent, got %d", len(results)) + } + + r := results[0] + if r.NumTurns != 2 { + t.Errorf("expected 2 turns, got %d", r.NumTurns) + } + if !r.IsError { + t.Error("expected IsError to be true for fallback result") + } + if r.Subtype != "stream_incomplete" { + t.Errorf("expected subtype stream_incomplete, got %q", r.Subtype) + } + // Input tokens: 5000 (turn 1) + 8000 (turn 2) = 13000 + if r.InputTokens != 13000 { + t.Errorf("expected 13000 input tokens, got %d", r.InputTokens) + } + // Output tokens: 2000 (turn 1) + 3000 (turn 2) = 5000 + if r.OutputTokens != 5000 { + t.Errorf("expected 5000 output tokens, got %d", r.OutputTokens) + } + // Cache read: 3000 + 4000 = 7000 + if r.CacheReadInputTokens != 7000 { + t.Errorf("expected 7000 cache read tokens, got %d", r.CacheReadInputTokens) + } + // Cache creation: 1000 + 0 = 1000 + if r.CacheCreationInputTokens != 1000 { + t.Errorf("expected 1000 cache creation tokens, got %d", r.CacheCreationInputTokens) + } + // TotalCostUSD cannot be derived without pricing data. + if r.TotalCostUSD != 0 { + t.Errorf("expected 0 total cost (not derivable), got %f", r.TotalCostUSD) + } +} + +// TestParseClaudeStreamNoFallbackWhenResultPresent verifies that the fallback +// ResultEvent is NOT emitted when a real result event is present. +func TestParseClaudeStreamNoFallbackWhenResultPresent(t *testing.T) { + lines := []string{ + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":5000,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}}}`, + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":2000}}}`, + `{"type":"result","num_turns":1,"total_cost_usd":0.05,"is_error":false,"subtype":"success","usage":{"input_tokens":5000,"output_tokens":2000,"cache_creation_input_tokens":0,"cache_read_input_tokens":0}}`, + } + + events := collectEvents(t, strings.Join(lines, "\n")) + + var results []ResultEvent + for _, e := range events { + if r, ok := e.(ResultEvent); ok { + results = append(results, r) + } + } + if len(results) != 1 { + t.Fatalf("expected exactly 1 ResultEvent (real, no fallback), got %d", len(results)) + } + if results[0].Subtype != "success" { + t.Errorf("expected subtype success from real result, got %q", results[0].Subtype) + } + if results[0].TotalCostUSD != 0.05 { + t.Errorf("expected 0.05 total cost from real result, got %f", results[0].TotalCostUSD) + } +} + +// TestProgressParserFallbackMetrics verifies that RunMetrics are populated +// from the fallback ResultEvent when the stream ends without a result line. +func TestProgressParserFallbackMetrics(t *testing.T) { + lines := []string{ + `{"type":"system","subtype":"init","model":"claude-opus-4-6","claude_code_version":"1.0.50"}`, + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":10000,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}}}`, + `{"type":"stream_event","event":{"type":"content_block_start","content_block":{"type":"tool_use","name":"Read"}}}`, + `{"type":"stream_event","event":{"type":"content_block_stop"}}`, + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":1500}}}`, + // No result event + } + + 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.NumTurns != 1 { + t.Errorf("expected 1 turn from fallback, got %d", metrics.NumTurns) + } + if metrics.InputTokens != 10000 { + t.Errorf("expected 10000 input tokens from fallback, got %d", metrics.InputTokens) + } + if metrics.OutputTokens != 1500 { + t.Errorf("expected 1500 output tokens from fallback, got %d", metrics.OutputTokens) + } + if metrics.Model != "claude-opus-4-6" { + t.Errorf("expected model claude-opus-4-6, got %q", metrics.Model) + } +} + +// TestParseClaudeStreamFallbackOnReadError verifies that the fallback +// ResultEvent is emitted when parseClaudeStream encounters a read error +// (e.g., broken pipe from a killed process). +func TestParseClaudeStreamFallbackOnReadError(t *testing.T) { + // Simulate a stream that ends with a read error after one turn. + normalData := strings.Join([]string{ + `{"type":"stream_event","event":{"type":"message_start","message":{"usage":{"input_tokens":7000,"cache_read_input_tokens":2000,"cache_creation_input_tokens":500}}}}`, + `{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":1000}}}`, + }, "\n") + "\n" + r := io.MultiReader( + strings.NewReader(normalData), + &errorReader{err: errors.New("broken pipe")}, + ) + + var events []AgentEvent + err := parseClaudeStream(r, func(evt AgentEvent) { + events = append(events, evt) + }) + if err == nil { + t.Fatal("expected error from parseClaudeStream, got nil") + } + + 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 fallback ResultEvent on read error, got %d", len(results)) + } + if results[0].NumTurns != 1 { + t.Errorf("expected 1 turn, got %d", results[0].NumTurns) + } + if results[0].InputTokens != 7000 { + t.Errorf("expected 7000 input tokens, got %d", results[0].InputTokens) + } + if results[0].OutputTokens != 1000 { + t.Errorf("expected 1000 output tokens, got %d", results[0].OutputTokens) + } +} + +// errorReader is a reader that always returns an error. +type errorReader struct { + err error +} + +func (e *errorReader) Read([]byte) (int, error) { + return 0, e.err +} + func TestHeartbeatConcurrency(t *testing.T) { var buf bytes.Buffer printer := ui.New(&buf) From 3ad460e603d511d31aa232ef94751ca34c3f8cc4 Mon Sep 17 00:00:00 2001 From: fullsend-fix <278716306+fullsend-ai-coder[bot]@users.noreply.github.com> Date: Mon, 31 Aug 2026 16:28:26 +0000 Subject: [PATCH 2/2] fix: set ErrorMessage on fallback ResultEvent Add ErrorMessage field to the synthetic ResultEvent emitted when the Claude stream ends without a result line, matching the pattern used by the real result handler and sibling parsers. Update fallback tests to assert the new field. Addresses review feedback on #6808 --- internal/runtime/claude_progress.go | 1 + internal/runtime/claude_progress_test.go | 7 +++++++ 2 files changed, 8 insertions(+) diff --git a/internal/runtime/claude_progress.go b/internal/runtime/claude_progress.go index e0aebf3246..02d68fd868 100644 --- a/internal/runtime/claude_progress.go +++ b/internal/runtime/claude_progress.go @@ -149,6 +149,7 @@ func parseClaudeStream(r io.Reader, onEvent func(AgentEvent)) error { onEvent(ResultEvent{ NumTurns: numTurns, IsError: true, + ErrorMessage: "stream ended without result event", Subtype: "stream_incomplete", InputTokens: cumulInput, OutputTokens: cumulOutput, diff --git a/internal/runtime/claude_progress_test.go b/internal/runtime/claude_progress_test.go index 795049825c..5e8806fcf7 100644 --- a/internal/runtime/claude_progress_test.go +++ b/internal/runtime/claude_progress_test.go @@ -698,6 +698,10 @@ func TestParseClaudeStreamFallbackResultOnIncompleteStream(t *testing.T) { if r.CacheCreationInputTokens != 1000 { t.Errorf("expected 1000 cache creation tokens, got %d", r.CacheCreationInputTokens) } + // ErrorMessage should describe the fallback condition. + if r.ErrorMessage != "stream ended without result event" { + t.Errorf("expected error message %q, got %q", "stream ended without result event", r.ErrorMessage) + } // TotalCostUSD cannot be derived without pricing data. if r.TotalCostUSD != 0 { t.Errorf("expected 0 total cost (not derivable), got %f", r.TotalCostUSD) @@ -807,6 +811,9 @@ func TestParseClaudeStreamFallbackOnReadError(t *testing.T) { if results[0].OutputTokens != 1000 { t.Errorf("expected 1000 output tokens, got %d", results[0].OutputTokens) } + if results[0].ErrorMessage != "stream ended without result event" { + t.Errorf("expected error message %q, got %q", "stream ended without result event", results[0].ErrorMessage) + } } // errorReader is a reader that always returns an error.