diff --git a/cmd/fullsend/main.go b/cmd/fullsend/main.go index cab80c6d51..076becf39b 100644 --- a/cmd/fullsend/main.go +++ b/cmd/fullsend/main.go @@ -18,9 +18,34 @@ type exitCoder interface { ExitCode() int } +// signalContext returns a context that is cancelled on the first +// SIGINT/SIGTERM, plus a cleanup function that stops signal forwarding. +// Subsequent signals are absorbed to prevent the default terminate handler +// from killing the process before cleanup (metrics, telemetry) completes. +// +// GitHub Actions sends SIGINT, waits ~7.5 s, then SIGTERM; without absorbing +// the second signal the default handler terminates the process before the +// metrics/telemetry flush path finishes (#6936). +// +// signal.NotifyContext stops listening after the first signal, which +// re-enables the default "terminate" handler for subsequent deliveries. +// Keeping our own channel registered prevents that. +func signalContext() (context.Context, func()) { + ctx, cancel := context.WithCancel(context.Background()) + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM) + go func() { + <-sigCh // First signal: cancel the context. + cancel() + for range sigCh { // Subsequent signals: absorbed. + } + }() + return ctx, func() { signal.Stop(sigCh) } +} + func main() { - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer stop() + ctx, cleanup := signalContext() + defer cleanup() if err := cli.Execute(ctx); err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) diff --git a/cmd/fullsend/main_test.go b/cmd/fullsend/main_test.go new file mode 100644 index 0000000000..b1e831b0cf --- /dev/null +++ b/cmd/fullsend/main_test.go @@ -0,0 +1,109 @@ +package main + +import ( + "os" + "os/signal" + "syscall" + "testing" + "time" +) + +// TestSignalContext_CancelsOnFirstSignal verifies that signalContext returns +// a context that is cancelled when the process receives the first SIGINT. +// This is the primary invariant for #6936: the first signal must cancel the +// context so the cleanup path (metrics, telemetry) can run. +func TestSignalContext_CancelsOnFirstSignal(t *testing.T) { + ctx, cleanup := signalContext() + defer cleanup() + + // Send SIGINT to ourselves. + proc, err := os.FindProcess(os.Getpid()) + if err != nil { + t.Fatalf("FindProcess: %v", err) + } + if err := proc.Signal(syscall.SIGINT); err != nil { + t.Fatalf("Signal: %v", err) + } + + select { + case <-ctx.Done(): + // Expected: context cancelled on first signal. + case <-time.After(2 * time.Second): + t.Fatal("context was not cancelled within 2s after SIGINT") + } +} + +// TestSignalContext_AbsorbsSubsequentSignals verifies that after the first +// signal cancels the context, further SIGINT/SIGTERM deliveries are absorbed +// (they do not kill the process). This is the second invariant for #6936: +// GitHub Actions sends SIGINT then SIGTERM ~7.5 s later; the second signal +// must not trigger the default terminate handler. +func TestSignalContext_AbsorbsSubsequentSignals(t *testing.T) { + ctx, cleanup := signalContext() + defer cleanup() + + proc, err := os.FindProcess(os.Getpid()) + if err != nil { + t.Fatalf("FindProcess: %v", err) + } + + // First signal: cancels the context. + if err := proc.Signal(syscall.SIGINT); err != nil { + t.Fatalf("first Signal: %v", err) + } + + select { + case <-ctx.Done(): + // Good — context cancelled. + case <-time.After(2 * time.Second): + t.Fatal("context was not cancelled after first SIGINT") + } + + // Second signal: must be absorbed (not kill the process). + if err := proc.Signal(syscall.SIGTERM); err != nil { + t.Fatalf("second Signal: %v", err) + } + + // If we reach this point without dying, the signal was absorbed. + // Give a short window for the signal to be delivered and handled. + time.Sleep(50 * time.Millisecond) +} + +// TestSignalContext_CleanupStopsForwarding verifies that after cleanup() is +// called, signal forwarding to signalContext's channel is disabled: a signal +// sent after cleanup() must not cancel ctx. A probe registered independently +// via signal.Notify confirms the signal was actually delivered by the OS +// (so the assertion isn't vacuously true because the signal never arrived), +// without relying on ctx or signalContext's own (now-detached) channel and +// without killing the test process. +func TestSignalContext_CleanupStopsForwarding(t *testing.T) { + ctx, cleanup := signalContext() + cleanup() + + probe := make(chan os.Signal, 1) + signal.Notify(probe, syscall.SIGINT) + defer signal.Stop(probe) + + proc, err := os.FindProcess(os.Getpid()) + if err != nil { + t.Fatalf("FindProcess: %v", err) + } + if err := proc.Signal(syscall.SIGINT); err != nil { + t.Fatalf("Signal: %v", err) + } + + select { + case <-probe: + // Expected: the OS still delivered SIGINT to an independent + // receiver, so the process is alive and the signal was sent. + case <-time.After(2 * time.Second): + t.Fatal("probe did not receive SIGINT within 2s; signal was never delivered") + } + + select { + case <-ctx.Done(): + t.Fatal("context was cancelled after cleanup(); signal.Stop did not disable forwarding") + default: + // Expected: forwarding disabled, context still active. + } +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 9c057e84c4..c9618014c7 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -2253,6 +2253,15 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep // Accumulate behavioral metrics across iterations. aggregateRunMetrics(&aggMetrics, &metrics, iteration) + if cancelled, cancelExitCode, cancelledErr := handleRunCancellation( + ctx, runErr, iteration, exitCode, rt.System(), rt.Name(), + &metrics, aggMetrics, runDir, agentSpan, attachIterationContent, + printer, lastIterElapsed, + ); cancelled { + lastExitCode = cancelExitCode + return cancelledErr + } + if runErr != nil { attachIterationContent("error") finalizeAgentSpan(agentSpan, runErr, iteration, exitCode, rt.System(), rt.Name(), &metrics, "") @@ -3676,6 +3685,52 @@ func transcriptErrorMessage(te agentruntime.TranscriptError) string { return truncateStatusMsgTo(te.DisplayMessage(), maxSpanEventMsgLen) } +// handleRunCancellation short-circuits the per-iteration loop in runAgent on +// context cancellation: it persists partial metrics and finalizes the agent +// span immediately, before extraction and validation that would be +// pointless on a dead sandbox. GitHub Actions cancellation (SIGTERM) +// terminates the process shortly after — writing metrics here ensures the +// artifact upload step (if: always()) captures the partial usage data +// (#6936). +// +// NOTE: TotalCostUSD will be zero in the persisted metrics because dollar +// cost is only available from the terminal ResultEvent, which a cancelled +// run never emits. Token counts (input, output, cache_read, cache_creation) +// are captured via the deferred TokensEvent and will be non-zero. See #6936 +// for background. +// +// cancelled is false when ctx is still live, in which case the caller's +// normal control flow continues unchanged; the other return values are +// meaningless in that case. +func handleRunCancellation( + ctx context.Context, + runErr error, + iteration, exitCode int, + system, runtimeName string, + metrics *agentruntime.RunMetrics, + aggMetrics aggregateMetrics, + runDir string, + agentSpan trace.Span, + attachIterationContent func(finishReason string), + printer *ui.Printer, + lastIterElapsed time.Duration, +) (cancelled bool, lastExitCode int, err error) { + cancelErr := ctx.Err() + if cancelErr == nil { + return false, 0, nil + } + if runErr == nil { + runErr = cancelErr + } + attachIterationContent("error") + finalizeAgentSpan(agentSpan, runErr, iteration, exitCode, system, runtimeName, metrics, "") + printer.StepWarn(fmt.Sprintf("Run cancelled (iteration %d, %.1fs elapsed)", iteration, lastIterElapsed.Seconds())) + if writeErr := writeMetricsJSON(runDir, aggMetrics); writeErr != nil { + printer.StepWarn("Failed to write metrics.json: " + writeErr.Error()) + } + return true, exitCode, fmt.Errorf("run cancelled (iteration %d): %w", iteration, runErr) +} + // finalizeAgentSpan records the end-of-iteration attributes and status on an // agent span and ends it. transcriptErr is non-empty when the transcript // reported a failure the process exit code did not (#2786): exit_code keeps diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index c4300cf503..e55b2c8223 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -30,6 +30,7 @@ import ( "github.com/fullsend-ai/fullsend/internal/harness" "github.com/fullsend-ai/fullsend/internal/mintclient" "github.com/fullsend-ai/fullsend/internal/resolve" + agentruntime "github.com/fullsend-ai/fullsend/internal/runtime" "github.com/fullsend-ai/fullsend/internal/ui" ) @@ -4853,6 +4854,241 @@ func TestWriteMetricsJSON(t *testing.T) { } } +// TestAggregateRunMetrics_PartialCancelledRun verifies that partial token +// metrics from a cancelled run (no ResultEvent, only TokensEvent) are folded +// into the aggregate correctly. This is the core data-flow assertion for #6936: +// the cancellation short-circuit writes metrics using the aggregate, so the +// aggregate must contain the partial tokens. +func TestAggregateRunMetrics_PartialCancelledRun(t *testing.T) { + var agg aggregateMetrics + + // Simulate a cancelled run: metrics populated via TokensEvent (no + // ResultEvent, so NumTurns/TotalCostUSD stay zero). + m := agentruntime.RunMetrics{ + InputTokens: 599, + OutputTokens: 119, + CacheCreationInputTokens: 148_943, + CacheReadInputTokens: 583_298, + Model: "claude-opus-4-6", + } + m.ToolCalls.Store(10) + + aggregateRunMetrics(&agg, &m, 1) + + if agg.TokenUsage.Input != 599 { + t.Errorf("token_usage.input = %d, want 599", agg.TokenUsage.Input) + } + if agg.TokenUsage.Output != 119 { + t.Errorf("token_usage.output = %d, want 119", agg.TokenUsage.Output) + } + if agg.TokenUsage.CacheCreation != 148_943 { + t.Errorf("token_usage.cache_creation = %d, want 148943", agg.TokenUsage.CacheCreation) + } + if agg.TokenUsage.CacheRead != 583_298 { + t.Errorf("token_usage.cache_read = %d, want 583298", agg.TokenUsage.CacheRead) + } + if agg.ToolCalls != 10 { + t.Errorf("tool_calls = %d, want 10", agg.ToolCalls) + } + if agg.Model != "claude-opus-4-6" { + t.Errorf("model = %q, want claude-opus-4-6", agg.Model) + } + if agg.NumTurns != 0 { + t.Errorf("num_turns = %d, want 0 (cancelled run has no ResultEvent)", agg.NumTurns) + } + if agg.TotalCostUSD != 0 { + t.Errorf("total_cost_usd = %f, want 0 (cancelled run has no ResultEvent)", agg.TotalCostUSD) + } +} + +// TestAggregateRunMetrics_MultiIterationCancel verifies that when a first +// iteration completes normally and the second is cancelled (partial tokens, +// no ResultEvent), the aggregate reflects both iterations' tokens and the +// cost from the successful iteration. This exercises the real-world +// cancellation scenario from #6936: the cleanup path writes the aggregate, +// so it must combine all iterations faithfully. +func TestAggregateRunMetrics_MultiIterationCancel(t *testing.T) { + var agg aggregateMetrics + + // Iteration 1: normal completion with a ResultEvent. + m1 := agentruntime.RunMetrics{ + InputTokens: 10_000, + OutputTokens: 2_000, + CacheCreationInputTokens: 50_000, + CacheReadInputTokens: 100_000, + NumTurns: 5, + TotalCostUSD: 0.42, + Model: "claude-opus-4-6", + } + m1.ToolCalls.Store(8) + aggregateRunMetrics(&agg, &m1, 1) + + // Iteration 2: cancelled — TokensEvent only (no ResultEvent). + m2 := agentruntime.RunMetrics{ + InputTokens: 599, + OutputTokens: 119, + CacheCreationInputTokens: 148_943, + CacheReadInputTokens: 583_298, + Model: "claude-opus-4-6", + } + m2.ToolCalls.Store(3) + aggregateRunMetrics(&agg, &m2, 2) + + // Token usage must reflect both iterations. + if agg.TokenUsage.Input != 10_599 { + t.Errorf("token_usage.input = %d, want 10599", agg.TokenUsage.Input) + } + if agg.TokenUsage.Output != 2_119 { + t.Errorf("token_usage.output = %d, want 2119", agg.TokenUsage.Output) + } + if agg.TokenUsage.CacheCreation != 198_943 { + t.Errorf("token_usage.cache_creation = %d, want 198943", agg.TokenUsage.CacheCreation) + } + if agg.TokenUsage.CacheRead != 683_298 { + t.Errorf("token_usage.cache_read = %d, want 683298", agg.TokenUsage.CacheRead) + } + + // Cost comes only from the successful iteration (cancelled run has zero cost). + if agg.TotalCostUSD != 0.42 { + t.Errorf("total_cost_usd = %f, want 0.42", agg.TotalCostUSD) + } + if agg.NumTurns != 5 { + t.Errorf("num_turns = %d, want 5 (cancelled iteration contributes zero turns)", agg.NumTurns) + } + if agg.ToolCalls != 11 { + t.Errorf("tool_calls = %d, want 11", agg.ToolCalls) + } + if agg.Iterations != 2 { + t.Errorf("iterations = %d, want 2", agg.Iterations) + } +} + +// TestWriteMetricsJSON_CancelledRunPartialTokens verifies that partial +// metrics from a cancelled run round-trip through writeMetricsJSON and +// contain the expected token values but zero cost. This is the persistence +// assertion for #6936: the artifact must contain non-zero token usage even +// when TotalCostUSD is unavailable. +func TestWriteMetricsJSON_CancelledRunPartialTokens(t *testing.T) { + dir := t.TempDir() + + // Build aggregate matching the cancelled-run evidence from #6936. + m := aggregateMetrics{ + Iterations: 1, + ToolCalls: 10, + Model: "claude-opus-4-6", + } + m.TokenUsage.Input = 599 + m.TokenUsage.Output = 119 + m.TokenUsage.CacheCreation = 148_943 + m.TokenUsage.CacheRead = 583_298 + + if err := writeMetricsJSON(dir, m); err != nil { + t.Fatalf("writeMetricsJSON failed: %v", err) + } + + data, err := os.ReadFile(filepath.Join(dir, metricsFile)) + if err != nil { + t.Fatalf("reading metrics.json: %v", err) + } + + var got aggregateMetrics + if err := json.Unmarshal(data, &got); err != nil { + t.Fatalf("unmarshalling metrics.json: %v", err) + } + + // Token usage must be non-zero — the primary assertion for #6936. + if got.TokenUsage.Input == 0 { + t.Error("expected non-zero token_usage.input in cancelled-run metrics") + } + if got.TokenUsage.Output == 0 { + t.Error("expected non-zero token_usage.output in cancelled-run metrics") + } + if got.TokenUsage.Input != 599 { + t.Errorf("token_usage.input = %d, want 599", got.TokenUsage.Input) + } + if got.TokenUsage.Output != 119 { + t.Errorf("token_usage.output = %d, want 119", got.TokenUsage.Output) + } + if got.TokenUsage.CacheCreation != 148_943 { + t.Errorf("token_usage.cache_creation = %d, want 148943", got.TokenUsage.CacheCreation) + } + if got.TokenUsage.CacheRead != 583_298 { + t.Errorf("token_usage.cache_read = %d, want 583298", got.TokenUsage.CacheRead) + } + if got.TotalCostUSD != 0 { + t.Errorf("total_cost_usd = %f, want 0 (dollar cost unavailable on cancellation)", got.TotalCostUSD) + } + if got.ToolCalls != 10 { + t.Errorf("tool_calls = %d, want 10", got.ToolCalls) + } +} + +// TestWriteMetricsJSON_MultiIterationCancelRoundTrip verifies the full +// data path for #6936: aggregate two iterations (one complete, one +// cancelled), write metrics.json, read it back, and verify the combined +// values survive serialization. This is the integration assertion — the +// cancellation short-circuit calls aggregateRunMetrics then writeMetricsJSON, +// so the round-trip must preserve both iterations' data. +func TestWriteMetricsJSON_MultiIterationCancelRoundTrip(t *testing.T) { + dir := t.TempDir() + + var agg aggregateMetrics + + // Iteration 1: complete. + m1 := agentruntime.RunMetrics{ + InputTokens: 10_000, + OutputTokens: 2_000, + CacheCreationInputTokens: 50_000, + CacheReadInputTokens: 100_000, + NumTurns: 5, + TotalCostUSD: 0.42, + Model: "claude-opus-4-6", + } + m1.ToolCalls.Store(8) + aggregateRunMetrics(&agg, &m1, 1) + + // Iteration 2: cancelled (partial tokens only). + m2 := agentruntime.RunMetrics{ + InputTokens: 599, + OutputTokens: 119, + Model: "claude-opus-4-6", + } + m2.ToolCalls.Store(3) + aggregateRunMetrics(&agg, &m2, 2) + + if err := writeMetricsJSON(dir, agg); err != nil { + t.Fatalf("writeMetricsJSON: %v", err) + } + + data, err := os.ReadFile(filepath.Join(dir, metricsFile)) + if err != nil { + t.Fatalf("reading metrics.json: %v", err) + } + + var got aggregateMetrics + if err := json.Unmarshal(data, &got); err != nil { + t.Fatalf("unmarshalling metrics.json: %v", err) + } + + // Combined token usage from both iterations. + if got.TokenUsage.Input != 10_599 { + t.Errorf("token_usage.input = %d, want 10599", got.TokenUsage.Input) + } + if got.TokenUsage.Output != 2_119 { + t.Errorf("token_usage.output = %d, want 2119", got.TokenUsage.Output) + } + // Cost from completed iteration only. + if got.TotalCostUSD != 0.42 { + t.Errorf("total_cost_usd = %f, want 0.42", got.TotalCostUSD) + } + if got.Iterations != 2 { + t.Errorf("iterations = %d, want 2", got.Iterations) + } + if got.ToolCalls != 11 { + t.Errorf("tool_calls = %d, want 11", got.ToolCalls) + } +} + // --- mintAgentToken tests --- // useZeroMintTokenBackoff overrides mintTokenBackoff to skip real sleeps so diff --git a/internal/cli/telemetry_run_test.go b/internal/cli/telemetry_run_test.go index 79f93f8272..6a6229955d 100644 --- a/internal/cli/telemetry_run_test.go +++ b/internal/cli/telemetry_run_test.go @@ -5,10 +5,12 @@ import ( "encoding/json" "errors" "fmt" + "io" "os" "path/filepath" "strings" "testing" + "time" "unicode/utf8" "github.com/stretchr/testify/assert" @@ -25,6 +27,7 @@ import ( agentruntime "github.com/fullsend-ai/fullsend/internal/runtime" "github.com/fullsend-ai/fullsend/internal/security" "github.com/fullsend-ai/fullsend/internal/telemetry" + "github.com/fullsend-ai/fullsend/internal/ui" ) func TestTelemetryExitCode(t *testing.T) { @@ -753,6 +756,120 @@ func TestFinalizeAgentSpan(t *testing.T) { }) } +// TestHandleRunCancellation exercises the cancellation short-circuit +// extracted from runAgent's per-iteration loop: the production path that +// persists partial metrics and finalizes the agent span when the run +// context is cancelled, before extraction and validation would otherwise +// run on a dead sandbox (#6936). This is the load-bearing branch a prior +// review iteration found untested — TestAggregateRunMetrics_* and +// TestWriteMetricsJSON_* only cover the helpers it calls, not the branch +// itself. +func TestHandleRunCancellation(t *testing.T) { + pinSpanLimitEnv(t) + + buildAgg := func() aggregateMetrics { + agg := aggregateMetrics{Iterations: 1, ToolCalls: 3, Model: "claude-opus-4-6"} + agg.TokenUsage.Input = 599 + agg.TokenUsage.Output = 119 + agg.TokenUsage.CacheCreation = 148_943 + agg.TokenUsage.CacheRead = 583_298 + return agg + } + + t.Run("cancelled context persists metrics, finalizes span, skips downstream", func(t *testing.T) { + runDir := t.TempDir() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + rec := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)) + _, span := tp.Tracer("test").Start(context.Background(), "agent") + + metrics := &agentruntime.RunMetrics{Model: "claude-opus-4-6"} + var attachedReasons []string + attach := func(reason string) { attachedReasons = append(attachedReasons, reason) } + printer := ui.New(io.Discard) + + cancelled, lastExitCode, err := handleRunCancellation( + ctx, nil, 2, 0, "anthropic", "claude", + metrics, buildAgg(), runDir, span, attach, printer, 1500*time.Millisecond, + ) + + require.True(t, cancelled, "ctx.Err() is non-nil: the short-circuit must fire") + assert.Equal(t, 0, lastExitCode) + require.Error(t, err) + assert.ErrorIs(t, err, context.Canceled, "the returned error must wrap the cancellation cause") + assert.Contains(t, err.Error(), "run cancelled (iteration 2)") + + assert.Equal(t, []string{"error"}, attachedReasons, "content must be attached with finish_reason=error before extraction/validation") + + ended := rec.Ended() + require.Len(t, ended, 1, "the agent span must be finalized (ended) on this path") + s := tracetest.SpanStubFromReadOnlySpan(ended[0]) + assert.Equal(t, codes.Error, s.Status.Code, "a cancelled iteration finalizes as an error status") + + data, readErr := os.ReadFile(filepath.Join(runDir, metricsFile)) + require.NoError(t, readErr, "metrics.json must be written before extraction/validation runs") + var got aggregateMetrics + require.NoError(t, json.Unmarshal(data, &got)) + assert.Equal(t, 599, got.TokenUsage.Input, "partial token counts must survive the write") + assert.Equal(t, 119, got.TokenUsage.Output) + assert.Equal(t, 148_943, got.TokenUsage.CacheCreation) + assert.Equal(t, 583_298, got.TokenUsage.CacheRead) + assert.Equal(t, float64(0), got.TotalCostUSD, "dollar cost is unavailable on cancellation") + }) + + t.Run("runErr already set is preserved as the wrapped cause", func(t *testing.T) { + runDir := t.TempDir() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + rec := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)) + _, span := tp.Tracer("test").Start(context.Background(), "agent") + + metrics := &agentruntime.RunMetrics{} + attach := func(string) {} + printer := ui.New(io.Discard) + runErr := errors.New("sandbox killed") + + cancelled, _, err := handleRunCancellation( + ctx, runErr, 1, -1, "anthropic", "claude", + metrics, buildAgg(), runDir, span, attach, printer, 0, + ) + + require.True(t, cancelled) + require.Error(t, err) + assert.Contains(t, err.Error(), "sandbox killed", "a non-nil runErr from rt.Run must not be discarded") + }) + + t.Run("live context does not short-circuit", func(t *testing.T) { + runDir := t.TempDir() + ctx := context.Background() // never cancelled + + rec := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)) + _, span := tp.Tracer("test").Start(context.Background(), "agent") + + metrics := &agentruntime.RunMetrics{} + attachCalled := false + attach := func(string) { attachCalled = true } + printer := ui.New(io.Discard) + + cancelled, _, err := handleRunCancellation( + ctx, nil, 1, 0, "anthropic", "claude", + metrics, buildAgg(), runDir, span, attach, printer, 0, + ) + + assert.False(t, cancelled, "a live context must not trigger the short-circuit") + assert.NoError(t, err) + assert.False(t, attachCalled, "content must not be attached when the run was not cancelled") + assert.Empty(t, rec.Ended(), "the span must not be finalized when the run was not cancelled") + _, statErr := os.Stat(filepath.Join(runDir, metricsFile)) + assert.True(t, os.IsNotExist(statErr), "metrics.json must not be written when the run was not cancelled") + }) +} + // pinSpanLimitEnv clears the OTEL span-limit variables so recorder tests // are hermetic — an ambient OTEL_SPAN_EVENT_COUNT_LIMIT=0 on a CI runner // would drop the exception events these tests assert on. diff --git a/internal/runtime/claude_progress_test.go b/internal/runtime/claude_progress_test.go index 0bd7b1a7e6..ca1c5f4f58 100644 --- a/internal/runtime/claude_progress_test.go +++ b/internal/runtime/claude_progress_test.go @@ -1239,6 +1239,61 @@ func TestParseClaudeStreamNoFinalTokensEventAfterResult(t *testing.T) { } } +// TestParseClaudeStreamBrokenPipeCapturesTokens verifies that when the +// stream is interrupted mid-read (broken pipe, simulating process kill), +// the deferred TokensEvent fires and metrics are captured. This is the +// regression test for #6936: GitHub Actions cancellation kills the sandbox +// subprocess, causing a broken pipe on the stream reader. +func TestParseClaudeStreamBrokenPipeCapturesTokens(t *testing.T) { + pr, pw := io.Pipe() + + // Write usage-bearing events then break the pipe (simulates SIGKILL). + go func() { + lines := []string{ + `{"type":"system","subtype":"init","model":"claude-opus-4-6"}`, + `{"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":2000}}}`, + } + for _, l := range lines { + pw.Write([]byte(l + "\n")) + } + pw.CloseWithError(errors.New("broken pipe")) + }() + + var metrics RunMetrics + var buf bytes.Buffer + printer := ui.New(&buf) + + // progressParser wraps parseClaudeStream and populates metrics. + err := progressParser(pr, printer, &metrics) + if err == nil { + t.Fatal("expected error from broken pipe, got nil") + } + + // Token metrics must be non-zero despite the broken pipe (#6936). + if metrics.InputTokens == 0 { + t.Error("expected non-zero InputTokens after broken pipe") + } + if metrics.OutputTokens == 0 { + t.Error("expected non-zero OutputTokens after broken pipe") + } + if metrics.InputTokens != 4000 { + t.Errorf("InputTokens = %d, want 4000", metrics.InputTokens) + } + if metrics.OutputTokens != 2000 { + t.Errorf("OutputTokens = %d, want 2000", metrics.OutputTokens) + } + if metrics.CacheReadInputTokens != 500 { + t.Errorf("CacheReadInputTokens = %d, want 500", metrics.CacheReadInputTokens) + } + if metrics.CacheCreationInputTokens != 200 { + t.Errorf("CacheCreationInputTokens = %d, want 200", metrics.CacheCreationInputTokens) + } + if metrics.Model != "claude-opus-4-6" { + t.Errorf("Model = %q, want claude-opus-4-6", metrics.Model) + } +} + // TestParseClaudeStreamFinalTokensEventOnCancel verifies that a deferred // TokensEvent is emitted at EOF when the stream ends without a ResultEvent, // even when the per-message total is below the throttle threshold.