From a194ce503223c8a45b59ba30efdd126990d25162 Mon Sep 17 00:00:00 2001 From: Dharit Shah Date: Mon, 29 Jun 2026 10:02:55 -0400 Subject: [PATCH 1/3] feat(#2577): add Level 1 distributed tracing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement Level 1 of ADR 0050: every `fullsend run` writes two zero-config, metadata-only telemetry artifacts to the run output dir — run-telemetry.jsonl (NDJSON span stream: run -> sandbox_create -> agent) and run-summary.json (aggregated summary with a W3C traceparent and a metrics block). Telemetry never affects the run: a nil/disabled recorder is a no-op, writes are recover()-guarded, the NDJSON is O_APPEND+Sync crash-safe, and the summary is written via atomic temp-file + rename. - New internal/telemetry package: W3C trace helpers + crash-safe Recorder. - Unified trace id: the per-run security trace id (dash-stripped) doubles as the W3C trace id, so one id ties security findings, telemetry, and child processes; TRACEPARENT is propagated to pre/post scripts. - fullsend.work_item_id correlation key on every span and the summary. - Span attrs use OTEL GenAI semconv names (gen_ai.request.model, gen_ai.system, gen_ai.usage.*) so the later L2 OTLP transform is ~1:1. - Metrics block reuses the existing aggregateMetrics (no new accounting): input/output/cache_creation/cache_read tokens, cost, turns, tool_calls, plus the resolved model. - Count tool_use blocks from Claude Code's nested message.content shape, and capture cache_read_input_tokens and the model from the stream. - Telemetry artifacts are metadata-only and excluded from the host output-redaction scan (the NDJSON is held open for append during it). Validated live in the rehearsal org end-to-end: tool_calls, cache_read, and model populate correctly. internal/telemetry coverage 92.5%. Signed-off-by: Dharit Shah --- internal/cli/run.go | 193 ++++++++-- internal/cli/scan_output_telemetry_test.go | 56 +++ internal/cli/telemetry_run_test.go | 195 ++++++++++ internal/runtime/claude_progress.go | 44 ++- internal/runtime/claude_progress_test.go | 82 +++- internal/runtime/runtime.go | 13 +- internal/telemetry/recorder.go | 318 ++++++++++++++++ internal/telemetry/recorder_test.go | 412 +++++++++++++++++++++ internal/telemetry/trace.go | 62 ++++ internal/telemetry/trace_test.go | 86 +++++ 10 files changed, 1428 insertions(+), 33 deletions(-) create mode 100644 internal/cli/scan_output_telemetry_test.go create mode 100644 internal/cli/telemetry_run_test.go create mode 100644 internal/telemetry/recorder.go create mode 100644 internal/telemetry/recorder_test.go create mode 100644 internal/telemetry/trace.go create mode 100644 internal/telemetry/trace_test.go diff --git a/internal/cli/run.go b/internal/cli/run.go index 93b9d2125..17b59a0bd 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "io" + "math" "net/http" "os" "os/exec" @@ -35,6 +36,7 @@ import ( "github.com/fullsend-ai/fullsend/internal/scaffold" "github.com/fullsend-ai/fullsend/internal/security" "github.com/fullsend-ai/fullsend/internal/statuscomment" + "github.com/fullsend-ai/fullsend/internal/telemetry" "github.com/fullsend-ai/fullsend/internal/ui" ) @@ -82,11 +84,14 @@ type aggregateMetrics struct { NumTurns int `json:"num_turns"` TotalCostUSD float64 `json:"total_cost_usd"` TokenUsage struct { - Input int `json:"input"` - Output int `json:"output"` + Input int `json:"input"` + Output int `json:"output"` + CacheCreation int `json:"cache_creation"` + CacheRead int `json:"cache_read"` } `json:"token_usage"` - Iterations int `json:"iterations"` - ToolCalls int `json:"tool_calls"` + Iterations int `json:"iterations"` + ToolCalls int `json:"tool_calls"` + Model string `json:"model,omitempty"` } func writeMetricsJSON(dir string, m aggregateMetrics) error { @@ -151,6 +156,10 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep printer.Header("Running agent: " + agentName) printer.Blank() + // runStart anchors the root telemetry span (ADR 0050); captured before any + // work so run-summary's total duration covers the whole invocation. + runStart := time.Now() + if rFlags.maxDepth < 0 { return fmt.Errorf("--max-depth must be >= 0, got %d", rFlags.maxDepth) } @@ -556,12 +565,21 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep } } + // Trace identity (ADR 0050 Level 1 + security correlation). Generated here, + // before the pre-script, so TRACEPARENT can propagate to child processes. + // The same id is reused as the security finding/audit trace id (dashed UUID) + // and, dash-stripped, as the W3C telemetry trace id — one id across both. + traceID := security.GenerateTraceID() + wTraceID := telemetry.TraceIDFromUUID(traceID) + rootSpanID := telemetry.NewSpanID() + workItemID := resolveWorkItemID() + // 2c. Run pre-script on the host (if configured). if h.PreScript != "" { preStart := time.Now() printer.StepStart("Running pre-script: " + h.PreScript) preCmd := exec.Command(h.PreScript) - preCmd.Env = append(os.Environ(), envToList(h.RunnerEnv)...) + preCmd.Env = childScriptEnv(h.RunnerEnv, telemetry.TraceParent(wTraceID, rootSpanID)) preCmd.Stdout = os.Stdout preCmd.Stderr = os.Stderr if err := preCmd.Run(); err != nil { @@ -573,18 +591,33 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep // 3. Create sandbox. sandboxName := fmt.Sprintf("agent-%s-%d-%d", agentName, os.Getpid(), time.Now().Unix()) + if outputBase == "" { + outputBase = filepath.Join(os.TempDir(), "fullsend") + } + runDir := filepath.Join(outputBase, sandboxName) + if err := os.MkdirAll(runDir, 0o755); err != nil { + return fmt.Errorf("creating run directory: %w", err) + } + + // Level 1 telemetry recorder (ADR 0050). A disabled recorder is a safe + // no-op, so any telemetry failure never affects the run. Finalize is + // registered before the post-script and cleanup defers so that — by LIFO + // order — it runs last and the summary captures the whole run. + var lastExitCode int + rec := telemetry.New(runDir, wTraceID, rootSpanID, agentName, workItemID, runStart) + defer func() { rec.Finalize(telemetryExitCode(lastExitCode, runErr)) }() + createStart := time.Now() printer.StepStart("Creating sandbox: " + sandboxName) + sandboxSpan := rec.StartSpan("sandbox_create", "", nil) readyTimeout := time.Duration(h.SandboxTimeoutSeconds) * time.Second if err := sandbox.CreateWithRetry(sandboxName, h.Providers, h.Image, h.Policy, sandbox.DefaultMaxCreateAttempts, readyTimeout); err != nil { + rec.EndSpan(sandboxSpan, "error", nil) printer.StepFail("Failed to create sandbox") return fmt.Errorf("creating sandbox: %w", err) } - if outputBase == "" { - outputBase = filepath.Join(os.TempDir(), "fullsend") - } - runDir := filepath.Join(outputBase, sandboxName) + rec.EndSpan(sandboxSpan, "ok", nil) // validationPassed is declared here (before the post-script defer) so the // defer closure can guard on it. The post-script must only run when @@ -616,7 +649,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep printer.StepStart("Running post-script: " + h.PostScript) postCmd := exec.Command(h.PostScript) postCmd.Dir = runDir - postCmd.Env = append(os.Environ(), envToList(h.RunnerEnv)...) + postCmd.Env = childScriptEnv(h.RunnerEnv, telemetry.TraceParent(wTraceID, rootSpanID)) postCmd.Stdout = os.Stdout postCmd.Stderr = os.Stderr if err := postCmd.Run(); err != nil { @@ -657,9 +690,6 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep repoName := filepath.Base(hostRepositoryDir) remoteRepositoryDir := fmt.Sprintf("%s/%s", sandbox.SandboxWorkspace, repoName) - // 5. Generate trace ID for security finding and audit log correlation. - traceID := security.GenerateTraceID() - // 6. Start runtime fetch service (Phase 4, ADR-0038). var fetchEnvVal fetchServiceEnv startFetch, deprecationWarning := shouldStartFetchService(h) @@ -869,10 +899,6 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep maxIterations = h.ValidationLoop.MaxIterations } - if err := os.MkdirAll(runDir, 0o755); err != nil { - return fmt.Errorf("creating run directory: %w", err) - } - oidcCtx, oidcCancel := context.WithCancel(context.Background()) var oidcWg sync.WaitGroup if oidcURL := os.Getenv("FULLSEND_GCP_OIDC_URL"); oidcURL != "" { @@ -893,7 +919,6 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep oidcWg.Wait() }() - var lastExitCode int var runCount int var aggMetrics aggregateMetrics @@ -928,6 +953,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep heartbeatDone := make(chan struct{}) go runHeartbeat(printer, agentStart, timeout, heartbeatDone) + agentSpan := rec.StartSpan("agent", "", map[string]any{"iteration": iteration}) var metrics agentruntime.RunMetrics exitCode, runErr := rt.Run(ctx, agentruntime.RunParams{ SandboxName: sandboxName, @@ -941,21 +967,28 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep }, printer, agentStart, &metrics) close(heartbeatDone) + agentStatus := "ok" + if runErr != nil { + agentStatus = "error" + } + rec.EndSpan(agentSpan, agentStatus, agentSpanEndAttrs(iteration, exitCode, &metrics)) + // Accumulate behavioral metrics across iterations. - aggMetrics.NumTurns += metrics.NumTurns - aggMetrics.TotalCostUSD += metrics.TotalCostUSD - aggMetrics.TokenUsage.Input += metrics.InputTokens - aggMetrics.TokenUsage.Output += metrics.OutputTokens - aggMetrics.ToolCalls += int(metrics.ToolCalls.Load()) - aggMetrics.Iterations = iteration + aggregateRunMetrics(&aggMetrics, &metrics, iteration) if runErr != nil { printer.StepFail("Agent execution failed") + // Record the real exit code (rt.Run returns -1 when the agent never + // started) so the telemetry summary reports the failure faithfully + // instead of collapsing every infra failure to a generic 1. + lastExitCode = exitCode // Write partial metrics before returning so downstream judges // (e.g., max_turns, max_cost) can inspect what happened. if err := writeMetricsJSON(runDir, aggMetrics); err != nil { printer.StepWarn("Failed to write metrics.json: " + err.Error()) } + rec.SetMetrics(toTelemetryMetrics(aggMetrics)) + rec.SetModel(aggMetrics.Model) return fmt.Errorf("running agent (iteration %d): %w", iteration, runErr) } lastExitCode = exitCode @@ -1051,6 +1084,8 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep if err := writeMetricsJSON(runDir, aggMetrics); err != nil { printer.StepWarn("Failed to write metrics.json: " + err.Error()) } + rec.SetMetrics(toTelemetryMetrics(aggMetrics)) + rec.SetModel(aggMetrics.Model) // 9e-bis. Surface transcript errors in workflow logs (GitHub Actions). // When the agent exits non-zero, parse transcript JSONL files and emit @@ -1508,6 +1543,106 @@ func validationFailMessage(output []byte, execErr error) string { } // envToList converts a map of env vars to a sorted list of KEY=VALUE strings. +// toTelemetryMetrics maps fullsend's aggregate run metrics onto the telemetry +// summary metrics — the same numbers already written to metrics.json, no new +// accounting. +// roundUSD rounds a dollar amount to cents for the telemetry output; +// metrics.json keeps full precision. +func roundUSD(c float64) float64 { return math.Round(c*100) / 100 } + +// agentSpanEndAttrs builds the span_end attributes for one agent iteration, +// using OTEL GenAI semconv names (gen_ai.*) so the later L2 OTLP transform is +// ~1:1. Cost is rounded to cents for telemetry; metrics.json keeps full precision. +func agentSpanEndAttrs(iteration, exitCode int, m *agentruntime.RunMetrics) map[string]any { + return map[string]any{ + "iteration": iteration, + "exit_code": exitCode, + "gen_ai.system": "anthropic", + "gen_ai.request.model": m.Model, + "gen_ai.usage.input_tokens": m.InputTokens, + "gen_ai.usage.output_tokens": m.OutputTokens, + "gen_ai.usage.cache_creation_input_tokens": m.CacheCreationInputTokens, + "gen_ai.usage.cache_read_input_tokens": m.CacheReadInputTokens, + "fullsend.cost_usd": roundUSD(m.TotalCostUSD), + "fullsend.tool_calls": int(m.ToolCalls.Load()), + } +} + +// aggregateRunMetrics folds one iteration's metrics into the cross-iteration +// aggregate: tokens/cost/turns/tool calls are summed; the last non-empty model +// wins. iteration records the highest iteration reached. +func aggregateRunMetrics(agg *aggregateMetrics, m *agentruntime.RunMetrics, iteration int) { + agg.NumTurns += m.NumTurns + agg.TotalCostUSD += m.TotalCostUSD + agg.TokenUsage.Input += m.InputTokens + agg.TokenUsage.Output += m.OutputTokens + agg.TokenUsage.CacheCreation += m.CacheCreationInputTokens + agg.TokenUsage.CacheRead += m.CacheReadInputTokens + agg.ToolCalls += int(m.ToolCalls.Load()) + agg.Iterations = iteration + if m.Model != "" { + agg.Model = m.Model + } +} + +func toTelemetryMetrics(m aggregateMetrics) telemetry.RunMetrics { + return telemetry.RunMetrics{ + InputTokens: m.TokenUsage.Input, + OutputTokens: m.TokenUsage.Output, + CacheCreationInputTokens: m.TokenUsage.CacheCreation, + CacheReadInputTokens: m.TokenUsage.CacheRead, + TotalCostUSD: roundUSD(m.TotalCostUSD), + NumTurns: m.NumTurns, + ToolCalls: m.ToolCalls, + } +} + +// resolveWorkItemID returns a stable cross-run correlation key for the work +// item being processed, sourced from existing run env (ADR 0049 leaves these +// context vars unchanged). The preference order yields a globally-unique, +// human-meaningful key; it falls back to "unknown" so Level 1's zero-config +// promise always holds. +func resolveWorkItemID() string { + if v := strings.TrimSpace(os.Getenv("ISSUE_KEY")); v != "" { + return v // source-neutral canonical key, e.g. "owner/repo#123" or "PROJ-123" + } + repo := strings.TrimSpace(os.Getenv("REPO_FULL_NAME")) + num := strings.TrimSpace(os.Getenv("ISSUE_NUMBER")) + if repo != "" && num != "" { + return repo + "#" + num + } + if v := strings.TrimSpace(os.Getenv("GITHUB_ISSUE_URL")); v != "" { + return v + } + if num != "" { + return num + } + return "unknown" +} + +// telemetryExitCode maps the run's final state to the exit code recorded in +// run-summary.json: the agent's last exit code, or 1 when the run failed for a +// non-agent reason (lastExitCode 0 with a non-nil error) so a failure is never +// reported as success. +func telemetryExitCode(lastExitCode int, runErr error) int { + if runErr != nil && lastExitCode == 0 { + return 1 + } + return lastExitCode +} + +// childScriptEnv builds the environment for a host-side child script (pre- or +// post-script): the harness RunnerEnv layered over the process environment, +// plus the W3C TRACEPARENT for trace propagation (ADR 0050 Level 1). An empty +// traceparent (telemetry disabled) is omitted rather than emitted blank. +func childScriptEnv(runnerEnv map[string]string, traceparent string) []string { + env := append(os.Environ(), envToList(runnerEnv)...) + if traceparent != "" { + env = append(env, "TRACEPARENT="+traceparent) + } + return env +} + func envToList(env map[string]string) []string { keys := make([]string, 0, len(env)) for k := range env { @@ -1942,6 +2077,16 @@ func scanOutputFiles(outputDir, traceID string, printer *ui.Printer) error { } return nil } + // Skip ONLY the recorder's own top-level telemetry artifacts (matched by + // exact path, not basename): they are metadata-only by construction, and + // run-telemetry.jsonl is still held open for append by the recorder during + // this scan — an in-place rewrite would truncate it under the open handle. + // Agent-written files that merely share these names (e.g. under an + // iteration output dir) are NOT exempt and must still be sanitized. + if path == filepath.Join(outputDir, telemetry.TelemetryFile) || + path == filepath.Join(outputDir, telemetry.SummaryFile) { + return nil + } content, readErr := os.ReadFile(path) if readErr != nil { relPath, _ := filepath.Rel(outputDir, path) diff --git a/internal/cli/scan_output_telemetry_test.go b/internal/cli/scan_output_telemetry_test.go new file mode 100644 index 000000000..e571b319c --- /dev/null +++ b/internal/cli/scan_output_telemetry_test.go @@ -0,0 +1,56 @@ +package cli + +import ( + "bytes" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/fullsend-ai/fullsend/internal/telemetry" + "github.com/fullsend-ai/fullsend/internal/ui" +) + +// TestScanOutputFilesSkipsTelemetryArtifacts pins that the host-side output +// redaction scan does NOT rewrite the telemetry files. The NDJSON file is still +// held open for append by the recorder during the scan, so an in-place rewrite +// would truncate it out from under the open handle; and both files are +// metadata-only by construction, so they don't need redaction. A normal output +// file must still be sanitized. +func TestScanOutputFilesSkipsTelemetryArtifacts(t *testing.T) { + dir := t.TempDir() + const secret = "Token: ghp_FAKEtesttoken000000000000000000000000\n" + + telem := filepath.Join(dir, telemetry.TelemetryFile) + summary := filepath.Join(dir, telemetry.SummaryFile) + normal := filepath.Join(dir, "output.txt") + // A sandbox agent could write a file that merely shares the telemetry name + // under its iteration output; that must still be sanitized (it is not the + // recorder's own artifact). + nested := filepath.Join(dir, "iteration-1", "output", telemetry.TelemetryFile) + require.NoError(t, os.MkdirAll(filepath.Dir(nested), 0o755)) + for _, p := range []string{telem, summary, normal, nested} { + require.NoError(t, os.WriteFile(p, []byte(secret), 0o644)) + } + + err := scanOutputFiles(dir, "trace-id", ui.New(&bytes.Buffer{})) + require.NoError(t, err) + + got, err := os.ReadFile(telem) + require.NoError(t, err) + assert.Equal(t, secret, string(got), "the recorder's own run-telemetry.jsonl must be left untouched") + + got, err = os.ReadFile(summary) + require.NoError(t, err) + assert.Equal(t, secret, string(got), "the recorder's own run-summary.json must be left untouched") + + got, err = os.ReadFile(normal) + require.NoError(t, err) + assert.NotContains(t, string(got), "ghp_FAKEtest", "non-telemetry output must still be sanitized") + + got, err = os.ReadFile(nested) + require.NoError(t, err) + assert.NotContains(t, string(got), "ghp_FAKEtest", "an agent file sharing the telemetry name must still be sanitized") +} diff --git a/internal/cli/telemetry_run_test.go b/internal/cli/telemetry_run_test.go new file mode 100644 index 000000000..f36d1e6be --- /dev/null +++ b/internal/cli/telemetry_run_test.go @@ -0,0 +1,195 @@ +package cli + +import ( + "fmt" + "regexp" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + agentruntime "github.com/fullsend-ai/fullsend/internal/runtime" + "github.com/fullsend-ai/fullsend/internal/security" + "github.com/fullsend-ai/fullsend/internal/telemetry" +) + +func TestTelemetryExitCode(t *testing.T) { + err := fmt.Errorf("boom") + assert.Equal(t, 0, telemetryExitCode(0, nil), "clean run => 0") + assert.Equal(t, 3, telemetryExitCode(3, nil), "agent exit code preserved on success") + // A non-agent failure (e.g. a later step errors) can leave lastExitCode==0; + // never report that as success. + assert.Equal(t, 1, telemetryExitCode(0, err), "lastExitCode 0 + error => 1, never success") + // The real agent infra-failure path: rt.Run returns (-1, err), and runAgent + // now records that -1 instead of masking it as 1. + assert.Equal(t, -1, telemetryExitCode(-1, err), "infra failure (-1) preserved faithfully") +} + +// TestTraceIDUnification pins the invariant runAgent relies on: the per-run +// security trace id (a dashed UUID, injected into the sandbox as +// FULLSEND_TRACE_ID) and the W3C telemetry trace id are the SAME underlying +// value — the telemetry id is just the security id with dashes stripped. This +// is what lets one id correlate security findings, telemetry, and child traces. +func TestTraceIDUnification(t *testing.T) { + uuid := security.GenerateTraceID() + require.True(t, security.IsValidTraceID(uuid), "security id must stay a valid dashed UUID for the sandbox") + + w := telemetry.TraceIDFromUUID(uuid) + assert.Equal(t, strings.ReplaceAll(uuid, "-", ""), w, "telemetry trace-id is the security id, dash-stripped") + assert.Regexp(t, regexp.MustCompile(`^[0-9a-f]{32}$`), w, "valid 32-hex W3C trace-id") +} + +func TestResolveWorkItemID(t *testing.T) { + cases := []struct { + name string + issueKey string + repoFull string + issueNumber string + issueURL string + want string + }{ + { + name: "ISSUE_KEY wins over everything", + issueKey: "PROJ-7", + repoFull: "octo/repo", + issueNumber: "9", + issueURL: "https://github.com/octo/repo/issues/9", + want: "PROJ-7", + }, + { + name: "repo + number forms canonical github key", + repoFull: "octo/repo", + issueNumber: "2577", + issueURL: "https://github.com/octo/repo/issues/2577", + want: "octo/repo#2577", + }, + { + name: "falls back to issue URL when repo missing", + issueURL: "https://github.com/octo/repo/issues/9", + want: "https://github.com/octo/repo/issues/9", + }, + { + name: "falls back to bare issue number", + issueNumber: "42", + want: "42", + }, + { + name: "unknown when nothing is set", + want: "unknown", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + // Set all four explicitly (empty == unset) to isolate from ambient env. + t.Setenv("ISSUE_KEY", tc.issueKey) + t.Setenv("REPO_FULL_NAME", tc.repoFull) + t.Setenv("ISSUE_NUMBER", tc.issueNumber) + t.Setenv("GITHUB_ISSUE_URL", tc.issueURL) + assert.Equal(t, tc.want, resolveWorkItemID()) + }) + } +} + +func TestChildScriptEnv_AppendsTraceparentOnce(t *testing.T) { + t.Setenv("FULLSEND_TEST_MARKER", "present") + const tp = "00-4f3a9c1b2d8e4a7c9f0b1e2d3c4a5b6d-a1b2c3d4e5f60718-01" + + env := childScriptEnv(map[string]string{"FOO": "bar"}, tp) + + traceparents, hasFoo, hasMarker := 0, false, false + for _, e := range env { + switch { + case strings.HasPrefix(e, "TRACEPARENT="): + traceparents++ + assert.Equal(t, "TRACEPARENT="+tp, e) + case e == "FOO=bar": + hasFoo = true + case e == "FULLSEND_TEST_MARKER=present": + hasMarker = true + } + } + assert.Equal(t, 1, traceparents, "exactly one TRACEPARENT entry") + assert.True(t, hasFoo, "RunnerEnv must be preserved") + assert.True(t, hasMarker, "process environment must be preserved") +} + +func TestChildScriptEnv_EmptyTraceparentOmitted(t *testing.T) { + env := childScriptEnv(map[string]string{"FOO": "bar"}, "") + for _, e := range env { + assert.False(t, strings.HasPrefix(e, "TRACEPARENT="), "no empty TRACEPARENT entry when disabled") + } +} + +func TestAgentSpanEndAttrs(t *testing.T) { + var m agentruntime.RunMetrics + m.Model = "claude-opus-4-6" + m.InputTokens = 11 + m.OutputTokens = 1505 + m.CacheCreationInputTokens = 38832 + m.CacheReadInputTokens = 109938 + m.TotalCostUSD = 0.335349 + m.ToolCalls.Store(11) + + a := agentSpanEndAttrs(2, 0, &m) + assert.Equal(t, 2, a["iteration"]) + assert.Equal(t, 0, a["exit_code"]) + assert.Equal(t, "anthropic", a["gen_ai.system"]) + assert.Equal(t, "claude-opus-4-6", a["gen_ai.request.model"]) + assert.Equal(t, 11, a["gen_ai.usage.input_tokens"]) + assert.Equal(t, 1505, a["gen_ai.usage.output_tokens"]) + assert.Equal(t, 38832, a["gen_ai.usage.cache_creation_input_tokens"]) + assert.Equal(t, 109938, a["gen_ai.usage.cache_read_input_tokens"]) + assert.Equal(t, 0.34, a["fullsend.cost_usd"], "cost rounded to cents") + assert.Equal(t, 11, a["fullsend.tool_calls"]) +} + +func TestAggregateRunMetrics(t *testing.T) { + var agg aggregateMetrics + + var m1 agentruntime.RunMetrics + m1.NumTurns, m1.TotalCostUSD = 5, 0.10 + m1.InputTokens, m1.OutputTokens = 10, 100 + m1.CacheCreationInputTokens, m1.CacheReadInputTokens = 1000, 5000 + m1.ToolCalls.Store(3) + m1.Model = "claude-opus-4-6" + aggregateRunMetrics(&agg, &m1, 1) + + var m2 agentruntime.RunMetrics // second iteration, no model reported + m2.NumTurns, m2.TotalCostUSD = 2, 0.05 + m2.InputTokens, m2.OutputTokens = 4, 40 + m2.CacheCreationInputTokens, m2.CacheReadInputTokens = 200, 900 + m2.ToolCalls.Store(2) + aggregateRunMetrics(&agg, &m2, 2) + + assert.Equal(t, 7, agg.NumTurns) + assert.InDelta(t, 0.15, agg.TotalCostUSD, 1e-9) + assert.Equal(t, 14, agg.TokenUsage.Input) + assert.Equal(t, 140, agg.TokenUsage.Output) + assert.Equal(t, 1200, agg.TokenUsage.CacheCreation) + assert.Equal(t, 5900, agg.TokenUsage.CacheRead) + assert.Equal(t, 5, agg.ToolCalls) + assert.Equal(t, 2, agg.Iterations) + assert.Equal(t, "claude-opus-4-6", agg.Model, "last non-empty model is retained") +} + +func TestToTelemetryMetrics(t *testing.T) { + var agg aggregateMetrics + agg.NumTurns = 7 + agg.TotalCostUSD = 0.24261625 + agg.TokenUsage.Input = 18432 + agg.TokenUsage.Output = 2901 + agg.TokenUsage.CacheCreation = 8000 + agg.TokenUsage.CacheRead = 50000 + agg.ToolCalls = 14 + agg.Iterations = 3 + + m := toTelemetryMetrics(agg) + assert.Equal(t, 18432, m.InputTokens, "input must map from TokenUsage.Input") + assert.Equal(t, 2901, m.OutputTokens, "output must map from TokenUsage.Output") + assert.Equal(t, 8000, m.CacheCreationInputTokens, "cache_creation must map from TokenUsage.CacheCreation") + assert.Equal(t, 50000, m.CacheReadInputTokens, "cache_read must map from TokenUsage.CacheRead") + assert.InDelta(t, 0.24, m.TotalCostUSD, 1e-9, "cost rounded to 2 decimals") + assert.Equal(t, 7, m.NumTurns) + assert.Equal(t, 14, m.ToolCalls) +} diff --git a/internal/runtime/claude_progress.go b/internal/runtime/claude_progress.go index 36ee54e5b..055aa8c18 100644 --- a/internal/runtime/claude_progress.go +++ b/internal/runtime/claude_progress.go @@ -24,9 +24,23 @@ type streamEvent struct { } // assistantMessage contains tool_use blocks from complete assistant messages. +// Claude Code's stream-json nests the content array (and model) under "message"; +// older/flat shapes put content at the top level. We accept both. type assistantMessage struct { Type string `json:"type"` Content json.RawMessage `json:"content"` + Message struct { + Content json.RawMessage `json:"content"` + Model string `json:"model"` + } `json:"message"` +} + +// systemEvent is Claude Code's initial "system"/"init" event, which carries the +// resolved model name. The result event does not include the model. +type systemEvent struct { + Type string `json:"type"` + Subtype string `json:"subtype"` + Model string `json:"model"` } type contentItem struct { @@ -55,8 +69,10 @@ type resultEvent struct { NumTurns int `json:"num_turns"` TotalCostUSD float64 `json:"total_cost_usd"` Usage struct { - InputTokens int `json:"input_tokens"` - OutputTokens int `json:"output_tokens"` + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens"` + CacheReadInputTokens int `json:"cache_read_input_tokens"` } `json:"usage"` } @@ -91,6 +107,13 @@ func progressParser(r io.Reader, printer *ui.Printer, start time.Time, metrics * continue } + if evt.Type == "system" { + var se systemEvent + if err := json.Unmarshal(line, &se); err == nil && se.Model != "" { + metrics.Model = se.Model + } + } + if evt.Type == "assistant" { parseAssistantToolUse(line, printer, start, metrics, isCI) } @@ -102,6 +125,8 @@ func progressParser(r io.Reader, printer *ui.Printer, start time.Time, metrics * metrics.TotalCostUSD = re.TotalCostUSD metrics.InputTokens = re.Usage.InputTokens metrics.OutputTokens = re.Usage.OutputTokens + metrics.CacheCreationInputTokens = re.Usage.CacheCreationInputTokens + metrics.CacheReadInputTokens = re.Usage.CacheReadInputTokens } } } @@ -113,8 +138,21 @@ func parseAssistantToolUse(line []byte, printer *ui.Printer, start time.Time, me return } + // Fall back to the assistant message's model when the system init event did + // not carry one, so gen_ai.request.model stays populated for all streams. + if metrics.Model == "" && msg.Message.Model != "" { + metrics.Model = msg.Message.Model + } + + // Real Claude Code output nests content under "message"; fall back to the + // top-level "content" for older/flat shapes. + content := msg.Message.Content + if len(content) == 0 { + content = msg.Content + } + var items []contentItem - if err := json.Unmarshal(msg.Content, &items); err != nil { + if err := json.Unmarshal(content, &items); err != nil { return } diff --git a/internal/runtime/claude_progress_test.go b/internal/runtime/claude_progress_test.go index 0c32cf67d..dc4f5491a 100644 --- a/internal/runtime/claude_progress_test.go +++ b/internal/runtime/claude_progress_test.go @@ -410,7 +410,7 @@ func TestProgressParserCapturesResultMetrics(t *testing.T) { lines := []string{ `{"type":"assistant","content":[{"type":"tool_use","name":"Read","input":{"file_path":"/src/main.go"}}]}`, `{"type":"assistant","content":[{"type":"tool_use","name":"Bash","input":{"command":"make test"}}]}`, - `{"type":"result","num_turns":8,"total_cost_usd":0.42,"usage":{"input_tokens":12000,"output_tokens":3400}}`, + `{"type":"result","num_turns":8,"total_cost_usd":0.42,"usage":{"input_tokens":12000,"output_tokens":3400,"cache_creation_input_tokens":8000,"cache_read_input_tokens":5000}}`, } input := strings.NewReader(strings.Join(lines, "\n")) @@ -437,6 +437,86 @@ func TestProgressParserCapturesResultMetrics(t *testing.T) { if metrics.OutputTokens != 3400 { t.Errorf("expected 3400 output tokens, got %d", metrics.OutputTokens) } + if metrics.CacheCreationInputTokens != 8000 { + t.Errorf("expected 8000 cache_creation input tokens, got %d", metrics.CacheCreationInputTokens) + } + if metrics.CacheReadInputTokens != 5000 { + t.Errorf("expected 5000 cache_read input tokens, got %d", metrics.CacheReadInputTokens) + } +} + +// TestProgressParserCountsToolCallsFromNestedMessage pins the real Claude Code +// stream-json shape: assistant events carry their content under "message", +// not at the top level. The earlier flat-shaped fixtures never exercised this, +// which let tool_use blocks go uncounted in production (every real run reported +// tool_calls: 0). See internal/runtime/claude_progress.go. +func TestProgressParserCountsToolCallsFromNestedMessage(t *testing.T) { + lines := []string{ + `{"type":"system","subtype":"init","model":"claude-opus-4-6"}`, + `{"type":"assistant","message":{"model":"claude-opus-4-6","content":[{"type":"text","text":"ok"},{"type":"tool_use","name":"Bash","input":{"command":"ls"}}]}}`, + `{"type":"assistant","message":{"model":"claude-opus-4-6","content":[{"type":"tool_use","name":"Read","input":{"file_path":"/a.go"}}]}}`, + `{"type":"assistant","message":{"model":"claude-opus-4-6","content":[{"type":"tool_use","name":"Skill","input":{}}]}}`, + `{"type":"result","num_turns":6}`, + } + + input := strings.NewReader(strings.Join(lines, "\n")) + var buf bytes.Buffer + printer := ui.New(&buf) + metrics := &RunMetrics{} + + if err := progressParser(input, printer, time.Now(), metrics); err != nil { + t.Fatalf("progressParser returned error: %v", err) + } + + if got := metrics.ToolCalls.Load(); got != 3 { + t.Errorf("expected 3 tool calls from nested message.content, got %d", got) + } +} + +// TestProgressParserCapturesModelFromAssistantWhenSystemLacksIt verifies the +// model falls back to the assistant message.model when the system init event +// carries no model, so gen_ai.request.model stays populated for all streams. +func TestProgressParserCapturesModelFromAssistantWhenSystemLacksIt(t *testing.T) { + lines := []string{ + `{"type":"system","subtype":"init"}`, // no model field + `{"type":"assistant","message":{"model":"claude-opus-4-6","content":[{"type":"text","text":"hi"}]}}`, + `{"type":"result","num_turns":1}`, + } + + input := strings.NewReader(strings.Join(lines, "\n")) + var buf bytes.Buffer + printer := ui.New(&buf) + metrics := &RunMetrics{} + + if err := progressParser(input, printer, time.Now(), metrics); err != nil { + t.Fatalf("progressParser returned error: %v", err) + } + + if metrics.Model != "claude-opus-4-6" { + t.Errorf("expected model from assistant message.model fallback, got %q", metrics.Model) + } +} + +// TestProgressParserCapturesModelFromSystemEvent verifies the resolved model is +// read from the system init event (the result event does not carry it). +func TestProgressParserCapturesModelFromSystemEvent(t *testing.T) { + lines := []string{ + `{"type":"system","subtype":"init","model":"claude-opus-4-6"}`, + `{"type":"result","num_turns":1}`, + } + + input := strings.NewReader(strings.Join(lines, "\n")) + var buf bytes.Buffer + printer := ui.New(&buf) + metrics := &RunMetrics{} + + if err := progressParser(input, printer, time.Now(), metrics); err != nil { + t.Fatalf("progressParser returned error: %v", err) + } + + if metrics.Model != "claude-opus-4-6" { + t.Errorf("expected model claude-opus-4-6 from system event, got %q", metrics.Model) + } } func TestProgressParserNoResultEvent(t *testing.T) { diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 5e5623382..100a9e160 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -10,11 +10,14 @@ import ( // RunMetrics collects execution statistics from stream parsing. type RunMetrics struct { - ToolCalls atomic.Int32 - NumTurns int `json:"num_turns"` - TotalCostUSD float64 `json:"total_cost_usd"` - InputTokens int `json:"input_tokens"` - OutputTokens int `json:"output_tokens"` + ToolCalls atomic.Int32 + NumTurns int `json:"num_turns"` + TotalCostUSD float64 `json:"total_cost_usd"` + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens"` + CacheReadInputTokens int `json:"cache_read_input_tokens"` + Model string `json:"model"` } // RunParams configures a single agent invocation inside the sandbox. diff --git a/internal/telemetry/recorder.go b/internal/telemetry/recorder.go new file mode 100644 index 000000000..1e1c0ed14 --- /dev/null +++ b/internal/telemetry/recorder.go @@ -0,0 +1,318 @@ +// Package telemetry implements fullsend's Level 1 distributed tracing: a +// zero-config, metadata-only local baseline (per ADR 0050). Every run writes a +// crash-safe NDJSON event stream (run-telemetry.jsonl) and an aggregated +// summary (run-summary.json) to the run output directory. +// +// The package is deliberately dependency-free and safe by construction: a nil +// *Recorder is valid, every method is a no-op when the recorder is nil or +// disabled, and no method returns an error. Any failure in the telemetry path +// disables the recorder rather than affecting the run — graceful degradation +// is a hard requirement. +package telemetry + +import ( + "encoding/json" + "os" + "path/filepath" + "sync" + "time" +) + +// SchemaVersion is the bespoke fullsend telemetry schema version, present on +// every event line and in the summary. Bump on incompatible schema changes. +const SchemaVersion = 1 + +// TelemetryFile and SummaryFile are the artifact names written to the output dir. +const ( + TelemetryFile = "run-telemetry.jsonl" + SummaryFile = "run-summary.json" +) + +// eventRecord is one NDJSON line in run-telemetry.jsonl. Metadata only — Attrs +// never contains prompt/completion content (that is Level 3). +type eventRecord struct { + V int `json:"v"` + Event string `json:"event"` // "span_start" | "span_end" + TraceID string `json:"trace_id"` + SpanID string `json:"span_id"` + Parent string `json:"parent"` + Name string `json:"name"` + TS string `json:"ts"` + WorkItemID string `json:"fullsend.work_item_id"` + DurationMS int64 `json:"duration_ms,omitempty"` // span_end only + Status string `json:"status,omitempty"` // span_end only + Attrs map[string]any `json:"attrs,omitempty"` +} + +// stepTiming is one entry in the summary's step list. +type stepTiming struct { + Name string `json:"name"` + DurationMS int64 `json:"duration_ms"` + Status string `json:"status"` +} + +// RunMetrics holds aggregate behavioral metrics for the run, surfaced into the +// summary via SetMetrics. Sourced from fullsend's existing per-run metrics (no +// new accounting). Omitted from the summary when never set. +type RunMetrics struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens"` + CacheReadInputTokens int `json:"cache_read_input_tokens"` + TotalCostUSD float64 `json:"total_cost_usd"` + NumTurns int `json:"num_turns"` + ToolCalls int `json:"tool_calls"` +} + +// runSummary is the content of run-summary.json, written once on Finalize. +type runSummary struct { + V int `json:"v"` + TraceID string `json:"trace_id"` + Traceparent string `json:"traceparent"` + Agent string `json:"agent"` + Model string `json:"model,omitempty"` + WorkItemID string `json:"fullsend.work_item_id"` + ExitCode int `json:"exit_code"` + StartedAt string `json:"started_at"` + EndedAt string `json:"ended_at"` + DurationMS int64 `json:"duration_ms"` + Steps []stepTiming `json:"steps"` + Metrics *RunMetrics `json:"metrics,omitempty"` +} + +type spanState struct { + name string + parent string + start time.Time +} + +// Recorder writes crash-safe NDJSON telemetry and a final summary. The zero +// value is not usable; obtain one from New. A nil *Recorder is a valid no-op. +type Recorder struct { + mu sync.Mutex + f *os.File + dir string + traceID string + rootSpanID string + agent string + model string + workItem string + start time.Time + spans map[string]*spanState + steps []stepTiming + metrics *RunMetrics + disabled bool + finalized bool +} + +// New opens /run-telemetry.jsonl for append and emits the root "run" span +// (backdated to start). traceID is a 32-hex W3C trace-id and rootSpanID a +// 16-hex span-id; both are supplied by the caller so the trace correlates with +// the run's security trace id and with child processes. +// +// New never returns an error: if the file cannot be opened it returns a +// disabled (no-op) recorder so the run is never affected. +func New(dir, traceID, rootSpanID, agent, workItemID string, start time.Time) *Recorder { + r := &Recorder{ + dir: dir, + traceID: traceID, + rootSpanID: rootSpanID, + agent: agent, + workItem: workItemID, + start: start, + spans: make(map[string]*spanState), + } + f, err := os.OpenFile(filepath.Join(dir, TelemetryFile), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + r.disabled = true + return r + } + r.f = f + + r.mu.Lock() + r.emit(eventRecord{ + V: SchemaVersion, Event: "span_start", TraceID: traceID, SpanID: rootSpanID, + Parent: "", Name: "run", TS: start.UTC().Format(time.RFC3339Nano), + WorkItemID: workItemID, Attrs: map[string]any{"agent": agent}, + }) + r.mu.Unlock() + return r +} + +// StartSpan emits a span_start and returns its span id. An empty parentID makes +// the span a child of the root span. Returns "" if the recorder is nil/disabled. +func (r *Recorder) StartSpan(name, parentID string, attrs map[string]any) string { + if r == nil { + return "" + } + r.mu.Lock() + defer r.mu.Unlock() + if r.disabled { + return "" + } + if parentID == "" { + parentID = r.rootSpanID + } + id := NewSpanID() + now := time.Now() + r.spans[id] = &spanState{name: name, parent: parentID, start: now} + r.emit(eventRecord{ + V: SchemaVersion, Event: "span_start", TraceID: r.traceID, SpanID: id, + Parent: parentID, Name: name, TS: now.UTC().Format(time.RFC3339Nano), + WorkItemID: r.workItem, Attrs: attrs, + }) + return id +} + +// EndSpan emits a span_end for spanID, records its duration as a summary step, +// and defaults an empty status to "ok". No-op for nil/disabled/empty spanID. +func (r *Recorder) EndSpan(spanID, status string, attrs map[string]any) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.disabled || spanID == "" { + return + } + now := time.Now() + name, parent := "", r.rootSpanID + var dur int64 + if st := r.spans[spanID]; st != nil { + name, parent = st.name, st.parent + dur = now.Sub(st.start).Milliseconds() + delete(r.spans, spanID) + } + if status == "" { + status = "ok" + } + r.steps = append(r.steps, stepTiming{Name: name, DurationMS: dur, Status: status}) + r.emit(eventRecord{ + V: SchemaVersion, Event: "span_end", TraceID: r.traceID, SpanID: spanID, + Parent: parent, Name: name, TS: now.UTC().Format(time.RFC3339Nano), + WorkItemID: r.workItem, DurationMS: dur, Status: status, Attrs: attrs, + }) +} + +// TraceParent returns the W3C traceparent for child-process propagation, or "" +// if the recorder is nil/disabled. +func (r *Recorder) TraceParent() string { + if r == nil { + return "" + } + r.mu.Lock() + defer r.mu.Unlock() + if r.disabled { + return "" + } + return TraceParent(r.traceID, r.rootSpanID) +} + +// SetMetrics records aggregate run metrics to include in run-summary.json. +// Safe to call repeatedly (last wins); no-op for nil/disabled recorders. +func (r *Recorder) SetMetrics(m RunMetrics) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.disabled { + return + } + r.metrics = &m +} + +// SetModel records the resolved model name to include in run-summary.json. +// Safe to call repeatedly (last wins); no-op for nil/disabled recorders. +func (r *Recorder) SetModel(model string) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.disabled { + return + } + r.model = model +} + +// Finalize emits the root span_end, writes run-summary.json atomically, and +// closes the telemetry file. It is idempotent and a no-op for nil/disabled. +func (r *Recorder) Finalize(exitCode int) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.finalized { + return + } + r.finalized = true + // Close the file if it was ever opened, even on the disabled path, so a + // mid-run emit failure (which sets r.disabled) doesn't leak the handle. + if r.f != nil { + defer func() { _ = r.f.Close() }() + } + if r.disabled { + return + } + + end := time.Now() + status := "ok" + if exitCode != 0 { + status = "error" + } + r.emit(eventRecord{ + V: SchemaVersion, Event: "span_end", TraceID: r.traceID, SpanID: r.rootSpanID, + Parent: "", Name: "run", TS: end.UTC().Format(time.RFC3339Nano), + WorkItemID: r.workItem, DurationMS: end.Sub(r.start).Milliseconds(), Status: status, + }) + + r.writeSummary(runSummary{ + V: SchemaVersion, TraceID: r.traceID, Traceparent: TraceParent(r.traceID, r.rootSpanID), + Agent: r.agent, Model: r.model, WorkItemID: r.workItem, ExitCode: exitCode, + StartedAt: r.start.UTC().Format(time.RFC3339Nano), EndedAt: end.UTC().Format(time.RFC3339Nano), + DurationMS: end.Sub(r.start).Milliseconds(), Steps: r.steps, Metrics: r.metrics, + }) +} + +// emit writes one JSON line + newline. The caller must hold r.mu and ensure the +// recorder is enabled. Any marshal/write failure or panic disables the recorder +// and is swallowed so the run is never affected. O_APPEND makes each write +// atomic at EOF; Sync keeps already-emitted lines durable across a crash. +func (r *Recorder) emit(rec eventRecord) { + defer func() { + if recover() != nil { + r.disabled = true + } + }() + data, err := json.Marshal(rec) + if err != nil { + r.disabled = true + return + } + if _, err := r.f.Write(append(data, '\n')); err != nil { + r.disabled = true + return + } + _ = r.f.Sync() +} + +// writeSummary writes run-summary.json atomically (temp file + rename) so a +// reader never observes a partial summary. Failures are swallowed and leave no +// stray temp file. The caller must hold r.mu. +func (r *Recorder) writeSummary(s runSummary) { + data, err := json.MarshalIndent(s, "", " ") + if err != nil { + return + } + tmp := filepath.Join(r.dir, SummaryFile+".tmp") + final := filepath.Join(r.dir, SummaryFile) + if err := os.WriteFile(tmp, append(data, '\n'), 0o644); err != nil { + _ = os.Remove(tmp) + return + } + if err := os.Rename(tmp, final); err != nil { + _ = os.Remove(tmp) + } +} diff --git a/internal/telemetry/recorder_test.go b/internal/telemetry/recorder_test.go new file mode 100644 index 000000000..f77269c97 --- /dev/null +++ b/internal/telemetry/recorder_test.go @@ -0,0 +1,412 @@ +package telemetry + +import ( + "bufio" + "encoding/json" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + testTraceID = "4f3a9c1b2d8e4a7c9f0b1e2d3c4a5b6d" + testRootID = "a1b2c3d4e5f60718" +) + +// readLines reads an NDJSON file and returns each non-empty line decoded as a +// map, failing the test if any complete line is not valid JSON. +func readLines(t *testing.T, path string) []map[string]any { + t.Helper() + f, err := os.Open(path) + require.NoError(t, err) + defer f.Close() + var out []map[string]any + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for sc.Scan() { + if strings.TrimSpace(sc.Text()) == "" { + continue + } + var m map[string]any + require.NoErrorf(t, json.Unmarshal(sc.Bytes(), &m), "line must be valid JSON: %s", sc.Text()) + out = append(out, m) + } + require.NoError(t, sc.Err()) + return out +} + +func TestRecorder_EmitsValidNDJSON(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "octo/repo#2577", time.Now()) + sp := r.StartSpan("sandbox_create", "", nil) + r.EndSpan(sp, "ok", nil) + r.Finalize(0) + + lines := readLines(t, filepath.Join(dir, "run-telemetry.jsonl")) + // root span_start, child span_start, child span_end, root span_end + require.Len(t, lines, 4) + for _, m := range lines { + assert.Equal(t, float64(SchemaVersion), m["v"]) + assert.Equal(t, testTraceID, m["trace_id"], "trace_id identical on every line") + assert.Equal(t, "octo/repo#2577", m["fullsend.work_item_id"], "work_item_id present on every line") + assert.NotEmpty(t, m["span_id"]) + assert.Contains(t, []any{"span_start", "span_end"}, m["event"]) + } +} + +func TestRecorder_RootSpanHasEmptyParentAndChildPointsToRoot(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + sp := r.StartSpan("sandbox_create", "", nil) // empty parent => defaults to root + r.EndSpan(sp, "ok", nil) + r.Finalize(0) + + lines := readLines(t, filepath.Join(dir, "run-telemetry.jsonl")) + for _, m := range lines { + if m["name"] == "run" { + assert.Equal(t, "", m["parent"], "root span has no parent") + } + if m["name"] == "sandbox_create" { + assert.Equal(t, testRootID, m["parent"], "empty parent defaults to root span id") + } + } +} + +func TestRecorder_SummaryFields(t *testing.T) { + dir := t.TempDir() + start := time.Now().Add(-2 * time.Second) + r := New(dir, testTraceID, testRootID, "code", "octo/repo#2577", start) + sp := r.StartSpan("agent", "", map[string]any{"iteration": 1}) + r.EndSpan(sp, "ok", map[string]any{"iteration": 1, "exit_code": 0}) + r.Finalize(0) + + data, err := os.ReadFile(filepath.Join(dir, "run-summary.json")) + require.NoError(t, err) + var s map[string]any + require.NoError(t, json.Unmarshal(data, &s)) + + assert.Equal(t, float64(SchemaVersion), s["v"]) + assert.Equal(t, testTraceID, s["trace_id"]) + assert.Equal(t, "code", s["agent"]) + assert.Equal(t, "octo/repo#2577", s["fullsend.work_item_id"]) + assert.Equal(t, float64(0), s["exit_code"]) + assert.NotEmpty(t, s["started_at"]) + assert.NotEmpty(t, s["ended_at"]) + durationMS, ok := s["duration_ms"].(float64) + require.True(t, ok) + assert.GreaterOrEqual(t, durationMS, float64(0)) + + tp, ok := s["traceparent"].(string) + require.True(t, ok) + assert.Regexp(t, reTraceparent, tp) + traceID, ok := s["trace_id"].(string) + require.True(t, ok) + assert.Contains(t, tp, traceID, "traceparent must embed trace_id") + + steps, ok := s["steps"].([]any) + require.True(t, ok) + require.Len(t, steps, 1) + step0, ok := steps[0].(map[string]any) + require.True(t, ok) + assert.Equal(t, "agent", step0["name"]) + assert.Equal(t, "ok", step0["status"]) + stepDur, ok := step0["duration_ms"].(float64) + require.True(t, ok) + assert.GreaterOrEqual(t, stepDur, float64(0)) +} + +func TestRecorder_NonZeroExitMarksError(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + r.Finalize(2) + + data, err := os.ReadFile(filepath.Join(dir, "run-summary.json")) + require.NoError(t, err) + var s map[string]any + require.NoError(t, json.Unmarshal(data, &s)) + assert.Equal(t, float64(2), s["exit_code"]) + + lines := readLines(t, filepath.Join(dir, "run-telemetry.jsonl")) + var rootEnd map[string]any + for _, m := range lines { + if m["name"] == "run" && m["event"] == "span_end" { + rootEnd = m + } + } + require.NotNil(t, rootEnd) + assert.Equal(t, "error", rootEnd["status"], "nonzero exit code => root span status error") +} + +func TestRecorder_CrashSafety_LinesDurableWithoutFinalize(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + sp := r.StartSpan("sandbox_create", "", nil) + r.EndSpan(sp, "ok", nil) + // Simulate a crash: Finalize never called. Synced lines must still parse. + lines := readLines(t, filepath.Join(dir, "run-telemetry.jsonl")) + require.Len(t, lines, 3, "root start + child start + child end persisted") + _, err := os.Stat(filepath.Join(dir, "run-summary.json")) + assert.True(t, os.IsNotExist(err), "no summary without Finalize") +} + +func TestRecorder_TruncatedTrailingLineTolerated(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "run-telemetry.jsonl") + content := `{"v":1,"event":"span_start","trace_id":"x","span_id":"a","name":"run"}` + "\n" + + `{"v":1,"event":"span_end","trace_id":"x","span_id":"a","name":"run"}` + "\n" + + `{"v":1,"event":"span_start","trace_id":"x","spa` // torn final line, no newline + require.NoError(t, os.WriteFile(path, []byte(content), 0o644)) + + f, err := os.Open(path) + require.NoError(t, err) + defer f.Close() + sc := bufio.NewScanner(f) + parsed := 0 + for sc.Scan() { + var m map[string]any + if json.Unmarshal(sc.Bytes(), &m) == nil { + parsed++ + } + } + assert.Equal(t, 2, parsed, "two complete lines parse; truncated final line is skipped") +} + +func TestRecorder_GracefulDegradation_UnwritableDir(t *testing.T) { + dir := filepath.Join(t.TempDir(), "does-not-exist") // parent missing => OpenFile fails + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + require.NotNil(t, r, "New must never return nil, even on failure") + + sp := r.StartSpan("x", "", nil) + assert.Equal(t, "", sp) + assert.NotPanics(t, func() { r.EndSpan(sp, "ok", nil) }) + assert.Equal(t, "", r.TraceParent(), "disabled recorder yields no traceparent") + assert.NotPanics(t, func() { r.Finalize(0) }) + + _, err := os.Stat(filepath.Join(dir, "run-summary.json")) + assert.Error(t, err, "nothing written when disabled") +} + +func TestRecorder_NilSafe(t *testing.T) { + var r *Recorder + assert.NotPanics(t, func() { + sp := r.StartSpan("x", "", nil) + r.EndSpan(sp, "ok", nil) + _ = r.TraceParent() + r.Finalize(0) + }) +} + +func TestRecorder_FinalizeIdempotentNoTmpLeft(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + r.Finalize(0) + assert.NotPanics(t, func() { r.Finalize(0) }, "second Finalize is a no-op") + + _, err := os.Stat(filepath.Join(dir, "run-summary.json.tmp")) + assert.True(t, os.IsNotExist(err), "no .tmp file should remain after atomic write") + _, err = os.Stat(filepath.Join(dir, "run-summary.json")) + assert.NoError(t, err) +} + +func TestRecorder_TraceParent(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + defer r.Finalize(0) + assert.Equal(t, "00-"+testTraceID+"-"+testRootID+"-01", r.TraceParent()) +} + +func TestRecorder_ConcurrentWritesNoCorruption(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + sp := r.StartSpan("agent", "", nil) + r.EndSpan(sp, "ok", nil) + }() + } + wg.Wait() + r.Finalize(0) + + // Mutex must prevent interleaved writes; every line stays valid JSON. + lines := readLines(t, filepath.Join(dir, "run-telemetry.jsonl")) + assert.Len(t, lines, 42, "root start + 20*(start+end) + root end") +} + +func TestRecorder_EndSpanUnknownSpanID(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + r.EndSpan("deadbeefdeadbeef", "ok", nil) // never started — must not panic + r.Finalize(0) + + lines := readLines(t, filepath.Join(dir, "run-telemetry.jsonl")) + var found map[string]any + for _, m := range lines { + if m["span_id"] == "deadbeefdeadbeef" { + found = m + } + } + require.NotNil(t, found) + assert.Equal(t, testRootID, found["parent"], "unknown span end defaults parent to root") + assert.Equal(t, "", found["name"]) +} + +func TestRecorder_WriteFailureMidRunDisablesGracefully(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + // Simulate the telemetry file failing mid-run by closing it underneath. + require.NoError(t, r.f.Close()) + + assert.NotPanics(t, func() { + sp := r.StartSpan("agent", "", nil) // emit hits a write error + r.EndSpan(sp, "ok", nil) + }) + assert.Equal(t, "", r.TraceParent(), "recorder disables itself after a write failure") + assert.NotPanics(t, func() { r.Finalize(0) }) +} + +func TestRecorder_SummaryWriteFailureSwallowed(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + // Point the summary target at a non-existent subdir so WriteFile fails; + // the failure must be swallowed and leave no stray temp file. + r.dir = filepath.Join(dir, "missing") + assert.NotPanics(t, func() { r.Finalize(0) }) + _, err := os.Stat(filepath.Join(r.dir, SummaryFile)) + assert.Error(t, err) + _, err = os.Stat(filepath.Join(r.dir, SummaryFile+".tmp")) + assert.True(t, os.IsNotExist(err), "no stray temp file on summary write failure") +} + +func TestRecorder_SummaryMetrics(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + r.SetMetrics(RunMetrics{InputTokens: 18432, OutputTokens: 2901, CacheCreationInputTokens: 8000, CacheReadInputTokens: 50000, TotalCostUSD: 0.0731, NumTurns: 7, ToolCalls: 14}) + r.Finalize(0) + + data, err := os.ReadFile(filepath.Join(dir, SummaryFile)) + require.NoError(t, err) + var s map[string]any + require.NoError(t, json.Unmarshal(data, &s)) + + m, ok := s["metrics"].(map[string]any) + require.True(t, ok, "summary must contain a metrics block") + assert.Equal(t, float64(18432), m["input_tokens"]) + assert.Equal(t, float64(2901), m["output_tokens"]) + assert.Equal(t, float64(8000), m["cache_creation_input_tokens"]) + assert.Equal(t, float64(50000), m["cache_read_input_tokens"]) + assert.InDelta(t, 0.0731, m["total_cost_usd"], 1e-9) + assert.Equal(t, float64(7), m["num_turns"]) + assert.Equal(t, float64(14), m["tool_calls"]) +} + +func TestRecorder_SummaryModel(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + r.SetModel("claude-opus-4-6") + r.Finalize(0) + + data, err := os.ReadFile(filepath.Join(dir, SummaryFile)) + require.NoError(t, err) + var s map[string]any + require.NoError(t, json.Unmarshal(data, &s)) + assert.Equal(t, "claude-opus-4-6", s["model"], "summary must carry the resolved model") +} + +func TestRecorder_SummaryModelOmittedWhenUnset(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + r.Finalize(0) // no SetModel + data, err := os.ReadFile(filepath.Join(dir, SummaryFile)) + require.NoError(t, err) + var s map[string]any + require.NoError(t, json.Unmarshal(data, &s)) + _, present := s["model"] + assert.False(t, present, "model must be omitted when never set") +} + +func TestRecorder_SetModelNilAndDisabledSafe(t *testing.T) { + var r *Recorder + assert.NotPanics(t, func() { r.SetModel("m") }) + disabled := New(filepath.Join(t.TempDir(), "does-not-exist"), testTraceID, testRootID, "code", "wi", time.Now()) + assert.NotPanics(t, func() { disabled.SetModel("m") }) +} + +func TestRecorder_FinalizeClosesFileEvenWhenDisabled(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + r.disabled = true // simulate a mid-run emit failure that disabled the recorder + r.Finalize(0) + _, err := r.f.Write([]byte("x")) + assert.ErrorIs(t, err, os.ErrClosed, "Finalize must close the open file even when disabled") +} + +func TestRecorder_EndSpanDefaultsEmptyStatusToOK(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + sp := r.StartSpan("sandbox_create", "", nil) + r.EndSpan(sp, "", nil) // empty status must default to "ok" + r.Finalize(0) + + data, err := os.ReadFile(filepath.Join(dir, SummaryFile)) + require.NoError(t, err) + var s map[string]any + require.NoError(t, json.Unmarshal(data, &s)) + steps := s["steps"].([]any) + require.Len(t, steps, 1) + assert.Equal(t, "ok", steps[0].(map[string]any)["status"], "empty status defaults to ok") +} + +func TestRecorder_EmitMarshalErrorDisablesRecorder(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + // A channel cannot be JSON-marshaled, so emit hits its marshal-error branch + // and must disable the recorder gracefully rather than panic. + assert.NotPanics(t, func() { + r.StartSpan("agent", "", map[string]any{"bad": make(chan int)}) + }) + assert.Equal(t, "", r.TraceParent(), "marshal failure disables the recorder") +} + +func TestRecorder_SummaryRenameFailureSwallowed(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + // Make the final summary path a non-empty directory so the atomic rename + // fails; the failure must be swallowed and leave no stray temp file. + blocker := filepath.Join(dir, SummaryFile) + require.NoError(t, os.MkdirAll(blocker, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(blocker, "keep"), []byte("x"), 0o644)) + + assert.NotPanics(t, func() { r.Finalize(0) }) + _, err := os.Stat(filepath.Join(dir, SummaryFile+".tmp")) + assert.True(t, os.IsNotExist(err), "no stray temp file when rename fails") +} + +func TestRecorder_SummaryMetricsOmittedWhenUnset(t *testing.T) { + dir := t.TempDir() + r := New(dir, testTraceID, testRootID, "code", "wi", time.Now()) + r.Finalize(0) // no SetMetrics + + data, err := os.ReadFile(filepath.Join(dir, SummaryFile)) + require.NoError(t, err) + var s map[string]any + require.NoError(t, json.Unmarshal(data, &s)) + _, present := s["metrics"] + assert.False(t, present, "metrics block must be omitted when never set") +} + +func TestRecorder_SetMetricsNilAndDisabledSafe(t *testing.T) { + var r *Recorder + assert.NotPanics(t, func() { r.SetMetrics(RunMetrics{InputTokens: 1}) }) + + disabled := New(filepath.Join(t.TempDir(), "does-not-exist"), testTraceID, testRootID, "code", "wi", time.Now()) + assert.NotPanics(t, func() { disabled.SetMetrics(RunMetrics{InputTokens: 1}) }) +} diff --git a/internal/telemetry/trace.go b/internal/telemetry/trace.go new file mode 100644 index 000000000..6061caaa7 --- /dev/null +++ b/internal/telemetry/trace.go @@ -0,0 +1,62 @@ +package telemetry + +import ( + "crypto/rand" + "encoding/hex" + "strings" +) + +// TraceIDFromUUID converts a dashed UUID (e.g. the output of +// security.GenerateTraceID) into a 32-hex-character W3C trace-id by removing +// the dashes. This deliberately reuses the per-run security trace id so a +// single id correlates security findings, telemetry, and child processes. +// +// W3C forbids an all-zero trace-id; if the input is malformed or strips to all +// zeros, a fresh random trace-id is returned instead so callers never emit an +// invalid value. +func TraceIDFromUUID(uuid string) string { + id := strings.ReplaceAll(uuid, "-", "") + if len(id) != 32 || id == "00000000000000000000000000000000" { + return randomHex(16) + } + return id +} + +// NewSpanID returns a random 16-hex-character (8-byte) W3C span-id. +func NewSpanID() string { + return randomHex(8) +} + +// TraceParent returns a W3C traceparent header value of the form +// "00-<32-hex trace-id>-<16-hex span-id>-01" (version 00, sampled flag 01). +func TraceParent(traceID, spanID string) string { + return "00-" + traceID + "-" + spanID + "-01" +} + +// randRead is a seam over crypto/rand.Read so the RNG-failure fallback in +// randomHex is testable. +var randRead = rand.Read + +// randomHex returns 2n lowercase hex characters from crypto/rand. If the RNG +// fails or (astronomically rarely) yields all-zero bytes, it falls back to a +// fixed non-zero pattern so the result is never an all-zero id — honoring the +// W3C invariant that trace/span ids must be non-zero. +func randomHex(n int) string { + b := make([]byte, n) + if _, err := randRead(b); err != nil || allZero(b) { + for i := range b { + b[i] = 0x11 + } + } + return hex.EncodeToString(b) +} + +// allZero reports whether every byte is zero. +func allZero(b []byte) bool { + for _, x := range b { + if x != 0 { + return false + } + } + return true +} diff --git a/internal/telemetry/trace_test.go b/internal/telemetry/trace_test.go new file mode 100644 index 000000000..c5fb6f75a --- /dev/null +++ b/internal/telemetry/trace_test.go @@ -0,0 +1,86 @@ +package telemetry + +import ( + "errors" + "regexp" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var ( + reTraceID = regexp.MustCompile(`^[0-9a-f]{32}$`) + reSpanID = regexp.MustCompile(`^[0-9a-f]{16}$`) + reTraceparent = regexp.MustCompile(`^00-[0-9a-f]{32}-[0-9a-f]{16}-01$`) +) + +const allZeroTraceID = "00000000000000000000000000000000" + +func TestTraceIDFromUUID_StripsDashes(t *testing.T) { + got := TraceIDFromUUID("4f3a9c1b-2d8e-4a7c-9f0b-1e2d3c4a5b6d") + assert.Equal(t, "4f3a9c1b2d8e4a7c9f0b1e2d3c4a5b6d", got) + assert.Regexp(t, reTraceID, got) +} + +func TestTraceIDFromUUID_GeneratorFormatIsValidW3C(t *testing.T) { + // Mirrors the shape of security.GenerateTraceID(): lowercase dashed UUIDv4. + got := TraceIDFromUUID("12ab34cd-56ef-4a7c-89b0-1e2d3c4a5b6d") + assert.Regexp(t, reTraceID, got, "must be 32 lowercase hex chars") + assert.NotEqual(t, allZeroTraceID, got) +} + +func TestTraceIDFromUUID_AllZeroIsRegenerated(t *testing.T) { + // W3C forbids an all-zero trace-id; the helper must not return one. + got := TraceIDFromUUID("00000000-0000-0000-0000-000000000000") + assert.Regexp(t, reTraceID, got) + assert.NotEqual(t, allZeroTraceID, got) +} + +func TestTraceIDFromUUID_MalformedIsReplaced(t *testing.T) { + got := TraceIDFromUUID("not-a-uuid") + assert.Regexp(t, reTraceID, got, "malformed input must still yield a valid 32-hex trace-id") +} + +func TestNewSpanID(t *testing.T) { + a := NewSpanID() + b := NewSpanID() + assert.Regexp(t, reSpanID, a) + assert.Regexp(t, reSpanID, b) + assert.NotEqual(t, "0000000000000000", a, "span-id must be non-zero") + assert.NotEqual(t, a, b, "span-ids must be unique across calls") +} + +func TestRandomHexFallsBackWhenRNGFails(t *testing.T) { + orig := randRead + randRead = func([]byte) (int, error) { return 0, errors.New("rng unavailable") } + defer func() { randRead = orig }() + + // 8 bytes of the 0x11 fallback pattern -> 16 "1" hex chars, never all-zero. + got := NewSpanID() + assert.Equal(t, "1111111111111111", got) + assert.Regexp(t, reSpanID, got) + assert.NotEqual(t, "0000000000000000", got) +} + +func TestRandomHexGuardsAllZeroFromRNG(t *testing.T) { + orig := randRead + // crypto/rand "succeeds" but yields all-zero bytes (astronomically rare): + // the result must still never be an all-zero id. + randRead = func(b []byte) (int, error) { + for i := range b { + b[i] = 0 + } + return len(b), nil + } + defer func() { randRead = orig }() + + assert.NotEqual(t, "0000000000000000", NewSpanID(), "all-zero RNG output must be replaced") + assert.Regexp(t, reSpanID, NewSpanID()) +} + +func TestTraceParent(t *testing.T) { + got := TraceParent("4f3a9c1b2d8e4a7c9f0b1e2d3c4a5b6d", "a1b2c3d4e5f60718") + require.Equal(t, "00-4f3a9c1b2d8e4a7c9f0b1e2d3c4a5b6d-a1b2c3d4e5f60718-01", got) + assert.Regexp(t, reTraceparent, got) +} From f30d09917da81a3f3a07b9e00496db008e38e00a Mon Sep 17 00:00:00 2001 From: Dharit Shah Date: Mon, 29 Jun 2026 16:11:02 -0400 Subject: [PATCH 2/3] refactor(#2577): source gen_ai.system from the runtime Address review feedback on the Level 1 telemetry PR: - Add Runtime.System() (Claude -> "anthropic") and thread it into the agent span attributes instead of hardcoding the vendor in the CLI, so telemetry stays runtime-agnostic per ADR 0050. - Note the per-event fsync as an L2 batching/async consideration. - Rename the scan test to the Test*_Scenario convention. Signed-off-by: Dharit Shah --- internal/cli/run.go | 9 +++++---- internal/cli/scan_output_telemetry_test.go | 2 +- internal/cli/telemetry_run_test.go | 4 ++-- internal/runtime/claude.go | 3 +++ internal/runtime/claude_test.go | 7 +++++++ internal/runtime/runtime.go | 4 ++++ internal/telemetry/recorder.go | 3 +++ 7 files changed, 25 insertions(+), 7 deletions(-) diff --git a/internal/cli/run.go b/internal/cli/run.go index 17b59a0bd..d4ac63224 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -971,7 +971,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep if runErr != nil { agentStatus = "error" } - rec.EndSpan(agentSpan, agentStatus, agentSpanEndAttrs(iteration, exitCode, &metrics)) + rec.EndSpan(agentSpan, agentStatus, agentSpanEndAttrs(iteration, exitCode, rt.System(), &metrics)) // Accumulate behavioral metrics across iterations. aggregateRunMetrics(&aggMetrics, &metrics, iteration) @@ -1552,12 +1552,13 @@ func roundUSD(c float64) float64 { return math.Round(c*100) / 100 } // agentSpanEndAttrs builds the span_end attributes for one agent iteration, // using OTEL GenAI semconv names (gen_ai.*) so the later L2 OTLP transform is -// ~1:1. Cost is rounded to cents for telemetry; metrics.json keeps full precision. -func agentSpanEndAttrs(iteration, exitCode int, m *agentruntime.RunMetrics) map[string]any { +// ~1:1. system is the runtime's gen_ai.system vendor (kept runtime-agnostic, not +// hardcoded). Cost is rounded to cents; metrics.json keeps full precision. +func agentSpanEndAttrs(iteration, exitCode int, system string, m *agentruntime.RunMetrics) map[string]any { return map[string]any{ "iteration": iteration, "exit_code": exitCode, - "gen_ai.system": "anthropic", + "gen_ai.system": system, "gen_ai.request.model": m.Model, "gen_ai.usage.input_tokens": m.InputTokens, "gen_ai.usage.output_tokens": m.OutputTokens, diff --git a/internal/cli/scan_output_telemetry_test.go b/internal/cli/scan_output_telemetry_test.go index e571b319c..45e503376 100644 --- a/internal/cli/scan_output_telemetry_test.go +++ b/internal/cli/scan_output_telemetry_test.go @@ -19,7 +19,7 @@ import ( // would truncate it out from under the open handle; and both files are // metadata-only by construction, so they don't need redaction. A normal output // file must still be sanitized. -func TestScanOutputFilesSkipsTelemetryArtifacts(t *testing.T) { +func TestScanOutputFiles_SkipsTelemetryArtifacts(t *testing.T) { dir := t.TempDir() const secret = "Token: ghp_FAKEtesttoken000000000000000000000000\n" diff --git a/internal/cli/telemetry_run_test.go b/internal/cli/telemetry_run_test.go index f36d1e6be..256388525 100644 --- a/internal/cli/telemetry_run_test.go +++ b/internal/cli/telemetry_run_test.go @@ -131,10 +131,10 @@ func TestAgentSpanEndAttrs(t *testing.T) { m.TotalCostUSD = 0.335349 m.ToolCalls.Store(11) - a := agentSpanEndAttrs(2, 0, &m) + a := agentSpanEndAttrs(2, 0, "anthropic", &m) assert.Equal(t, 2, a["iteration"]) assert.Equal(t, 0, a["exit_code"]) - assert.Equal(t, "anthropic", a["gen_ai.system"]) + assert.Equal(t, "anthropic", a["gen_ai.system"], "gen_ai.system is sourced from the runtime, not hardcoded") assert.Equal(t, "claude-opus-4-6", a["gen_ai.request.model"]) assert.Equal(t, 11, a["gen_ai.usage.input_tokens"]) assert.Equal(t, 1505, a["gen_ai.usage.output_tokens"]) diff --git a/internal/runtime/claude.go b/internal/runtime/claude.go index 5b0b701de..0c83d51f1 100644 --- a/internal/runtime/claude.go +++ b/internal/runtime/claude.go @@ -23,6 +23,9 @@ type ClaudeRuntime struct{} func (ClaudeRuntime) Name() string { return "claude" } +// System returns the OTEL GenAI `gen_ai.system` vendor for Claude Code's models. +func (ClaudeRuntime) System() string { return "anthropic" } + func (ClaudeRuntime) ConfigDir() string { return sandbox.SandboxClaudeConfig } func (ClaudeRuntime) WorkspaceDir() string { return sandbox.SandboxWorkspace } diff --git a/internal/runtime/claude_test.go b/internal/runtime/claude_test.go index 9aab87183..942949021 100644 --- a/internal/runtime/claude_test.go +++ b/internal/runtime/claude_test.go @@ -475,3 +475,10 @@ func TestResolveSkillDisplayName(t *testing.T) { }) } } + +func TestClaudeRuntimeSystem(t *testing.T) { + // gen_ai.system is the OTEL vendor value for the runtime's models; Claude + // Code runs Anthropic models. Sourcing it from the runtime (not hardcoding + // in the CLI) keeps telemetry runtime-agnostic per ADR 0050. + assert.Equal(t, "anthropic", ClaudeRuntime{}.System()) +} diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 100a9e160..1342cfc7e 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -43,6 +43,10 @@ type TranscriptError struct { // Runtime is an agent execution backend (LLM tool-use loop) inside the sandbox. type Runtime interface { Name() string + // System returns the OTEL GenAI `gen_ai.system` value (the model vendor) for + // this runtime, e.g. "anthropic". Kept on the runtime so telemetry stays + // runtime-agnostic rather than hardcoding a vendor in the CLI (ADR 0050). + System() string ConfigDir() string WorkspaceDir() string EnvExports() []string diff --git a/internal/telemetry/recorder.go b/internal/telemetry/recorder.go index 1e1c0ed14..e2c4bb6fd 100644 --- a/internal/telemetry/recorder.go +++ b/internal/telemetry/recorder.go @@ -295,6 +295,9 @@ func (r *Recorder) emit(rec eventRecord) { r.disabled = true return } + // fsync per event is fine for L1's handful of spans per run; if L2/L3 raises + // span volume, batch or move to an async flush (fsync can be 5-15ms on slow/ + // network filesystems). _ = r.f.Sync() } From 87cecd8b4fbcc2f48529e237231187b31c8efc1c Mon Sep 17 00:00:00 2001 From: Dharit Shah Date: Tue, 30 Jun 2026 09:45:31 -0400 Subject: [PATCH 3/3] refactor(#2577): rename traceID to securityTraceID Disambiguate the security trace id (dashed UUID) from the W3C telemetry trace id (wTraceID) in runAgent, per review feedback. Variable rename only; the helper-function parameters keep their generic traceID name. Signed-off-by: Dharit Shah --- internal/cli/run.go | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/internal/cli/run.go b/internal/cli/run.go index d4ac63224..9e50f33b4 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -569,8 +569,8 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep // before the pre-script, so TRACEPARENT can propagate to child processes. // The same id is reused as the security finding/audit trace id (dashed UUID) // and, dash-stripped, as the W3C telemetry trace id — one id across both. - traceID := security.GenerateTraceID() - wTraceID := telemetry.TraceIDFromUUID(traceID) + securityTraceID := security.GenerateTraceID() + wTraceID := telemetry.TraceIDFromUUID(securityTraceID) rootSpanID := telemetry.NewSpanID() workItemID := resolveWorkItemID() @@ -702,7 +702,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep FetchPolicy: fetch.DefaultPolicy, WorkspaceRoot: absFullsendDir, AuditLogPath: filepath.Join(absFullsendDir, ".fullsend-cache", "fetch-audit.jsonl"), - TraceID: traceID, + TraceID: securityTraceID, SandboxName: sandboxName, MaxFetches: h.EffectiveMaxRuntimeFetches(), Uploader: &fetchsvc.SandboxUploader{}, @@ -830,8 +830,8 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep } // 9a. Display trace ID (generated earlier for fetch service audit logging). - printer.KeyValue("Trace ID", traceID) - if err := injectTraceID(sandboxName, traceID); err != nil { + printer.KeyValue("Trace ID", securityTraceID) + if err := injectTraceID(sandboxName, securityTraceID); err != nil { printer.StepWarn("Could not inject trace ID into sandbox: " + err.Error()) } @@ -840,7 +840,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep // SKILL.md) that were just copied into the sandbox. if h.SecurityEnabled() { printer.StepStart("Running pre-agent security scan") - scanCmd := buildScanContextCommand(remoteRepositoryDir, traceID) + scanCmd := buildScanContextCommand(remoteRepositoryDir, securityTraceID) stdout, stderr, exitCode, execErr := sandbox.Exec(sandboxName, scanCmd, 60*time.Second) if execErr != nil { printer.StepFail("Security scan failed: " + execErr.Error()) @@ -1103,7 +1103,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep // 9f. Post-agent output scan — redact secrets from extracted output. if h.SecurityEnabled() { printer.StepStart("Running post-agent output scan") - if err := scanOutputFiles(runDir, traceID, printer); err != nil { + if err := scanOutputFiles(runDir, securityTraceID, printer); err != nil { printer.StepWarn("Output scan error: " + err.Error()) } @@ -1138,7 +1138,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep printer.KeyValue("Run directory", runDir) printer.KeyValue("Agent exit code", fmt.Sprintf("%d", lastExitCode)) printer.KeyValue("Agent runs", fmt.Sprintf("%d", runCount)) - printer.KeyValue("Trace ID", traceID) + printer.KeyValue("Trace ID", securityTraceID) if h.ValidationLoop != nil { if validationPassed { printer.KeyValue("Validation", "passed")