diff --git a/internal/runtime/claude_progress.go b/internal/runtime/claude_progress.go index bfab70e6ff..02d68fd868 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,44 @@ 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, + ErrorMessage: "stream ended without result event", + 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 +273,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 +310,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..5e8806fcf7 100644 --- a/internal/runtime/claude_progress_test.go +++ b/internal/runtime/claude_progress_test.go @@ -633,6 +633,198 @@ 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) + } + // 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) + } +} + +// 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) + } + 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. +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)