diff --git a/docs/ADRs/0050-distributed-tracing-instrumentation.md b/docs/ADRs/0050-distributed-tracing-instrumentation.md index c004e115b1..7dc9f386b9 100644 --- a/docs/ADRs/0050-distributed-tracing-instrumentation.md +++ b/docs/ADRs/0050-distributed-tracing-instrumentation.md @@ -161,8 +161,10 @@ online scoring of wild-run traces writes `eval-measurements.jsonl` beside telemetry when at least one new score is produced (tool-agnostic). Distinct from functional eval fixtures ([ADR 0051](0051-agent-eval-harness-for-test-infrastructure.md)). -> **Planned:** portable remote score export follows the same OTLP -> configuration as this ADR — no vendor score adapters in core. +> **Done ([#6459](https://github.com/fullsend-ai/fullsend/pull/6459) / +> [ADR 0087](0087-eval-measurements-online-trace-scoring.md)):** portable +> remote score export uses the same OTLP configuration as this ADR — no +> vendor score adapters in core. **2026-08-18 — Remove duplicate token/cost from root span (3278b059):** `gen_ai.request.model` and `gen_ai.usage.*` token attributes moved to agent diff --git a/docs/ADRs/0087-eval-measurements-online-trace-scoring.md b/docs/ADRs/0087-eval-measurements-online-trace-scoring.md index ceff7a86e6..fc92d22f03 100644 --- a/docs/ADRs/0087-eval-measurements-online-trace-scoring.md +++ b/docs/ADRs/0087-eval-measurements-online-trace-scoring.md @@ -99,6 +99,11 @@ Until that release lands, GHA/GitLab `eval-measure` wiring is provisional (clean skip when the remote manifest is missing). Local `FULLSEND_DIR` manifests are exercised in unit tests today. +> **Implemented ([#6459](https://github.com/fullsend-ai/fullsend/pull/6459)):** +> portable remote score export now ships via the shared OTEL path as +> `gen_ai.evaluation.result` span events (fail-open). Decision text above +> is unchanged; this note records delivery only. + The first scorer is `trace_fitness` (catalog id `em-001`) — span-tree and attribute fitness so later scorers can trust the trace. EM-001 reads OpenTelemetry GenAI attribute names (`gen_ai.*` constants in diff --git a/docs/architecture.md b/docs/architecture.md index 6f61f3ae8f..900a3a149b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -352,15 +352,13 @@ Observability is a cross-cutting concern that touches every other component. Eac - JSONL reasoning trace exposure: raw JSONL conversation transcripts are extracted from sandboxes and stored with owner-scoped access. Credential scanning acts as an invariant check on [ADR 0017](ADRs/0017-credential-isolation-for-sandboxed-agents.md)'s isolation model. Agents handling data from protected sources beyond the target repo can opt in to JSONL suppression via configuration ([ADR 0021](ADRs/0021-jsonl-reasoning-trace-exposure.md)). - Event-driven stage dispatch remains traceable end-to-end in the GitHub Actions UI by using synchronous `workflow_call` dispatch (see [ADR 0041](ADRs/0041-synchronous-workflow-call-event-dispatch.md)). - Distributed tracing: framework-native OpenTelemetry instrumentation with zero-configuration baseline. Every run produces `run-telemetry.jsonl` locally; optional live OTLP export to any compatible backend. W3C trace context propagation links multi-agent pipelines into unified traces. OTEL GenAI semantic conventions enable LLM-aware backends ([ADR 0050](ADRs/0050-distributed-tracing-instrumentation.md)). -- Eval measurements: the concept of scoring traces ([fail-open](glossary.md#fail-open)). [OTEL primary facts](glossary.md#otel-primary-facts) stay on the run trace (`run-telemetry.jsonl`); [OTEL derived products](glossary.md#otel-derived-products) are the scores (`eval-measurements.jsonl`) ([ADR 0087](ADRs/0087-eval-measurements-online-trace-scoring.md)). See [Eval Measurements](guides/infrastructure/eval-measurements.md). - - > **Planned:** portable remote score export via the same OTLP configuration as agent traces ([ADR 0087](ADRs/0087-eval-measurements-online-trace-scoring.md)). Not yet implemented. +- Eval measurements: the concept of scoring traces ([fail-open](glossary.md#fail-open)). [OTEL primary facts](glossary.md#otel-primary-facts) stay on the run trace (`run-telemetry.jsonl`); [OTEL derived products](glossary.md#otel-derived-products) are the scores (`eval-measurements.jsonl`) ([ADR 0087](ADRs/0087-eval-measurements-online-trace-scoring.md)). See [Eval Measurements](guides/infrastructure/eval-measurements.md). When `OTEL_EXPORTER_OTLP_*` is set, scores also export as `gen_ai.evaluation.result` span events on the same TraceID (same OTLP path as agent traces; fail-open). **Open questions:** - What signals matter most — cost, latency, token usage, action logs, decision traces, or something else? - ~~How do we balance detailed tracing (useful for debugging) with the volume of data agents will produce?~~ Decided in [ADR 0050](ADRs/0050-distributed-tracing-instrumentation.md): instrument all lifecycle steps comprehensively; volume is managed by backends not by suppressing data at the source. -- ~~How do we score wild agent traces for trends without a second export stack?~~ Decided in [ADR 0087](ADRs/0087-eval-measurements-online-trace-scoring.md): eval measurements write local JSONL beside telemetry when at least one new score row is produced (including `label: skip`); portable remote export uses the same OTLP config as traces (planned). The JSONL is absent (not empty) when telemetry/manifest is missing, no traces match, or every candidate is already in the ledger. +- ~~How do we score wild agent traces for trends without a second export stack?~~ Decided in [ADR 0087](ADRs/0087-eval-measurements-online-trace-scoring.md): eval measurements write local JSONL beside telemetry when at least one new score row is produced (including `label: skip`); portable remote export uses the same OTLP config as traces (`gen_ai.evaluation.result` events). The JSONL is absent (not empty) when telemetry/manifest is missing, no traces match, or every candidate is already in the ledger. - What is the retention and access model for agent logs? Who can see what? (JSONL trace access model decided in [ADR 0021](ADRs/0021-jsonl-reasoning-trace-exposure.md); retention policy and broader log access remain open.) - How does observability interact with the security requirement that "every action is logged, attributable, and reviewable"? (See [security-threat-model.md](problems/security-threat-model.md).) - Is there a real-time monitoring requirement (agent is stuck, agent is behaving anomalously), or is observability primarily forensic? diff --git a/docs/guides/infrastructure/distributed-tracing.md b/docs/guides/infrastructure/distributed-tracing.md index 1a6512ece6..060e9748fd 100644 --- a/docs/guides/infrastructure/distributed-tracing.md +++ b/docs/guides/infrastructure/distributed-tracing.md @@ -293,9 +293,9 @@ authentication mechanism. After each managed agent run, `fullsend eval-measure` scores `run-telemetry.jsonl` in the same job (fail-open). Scores land in `eval-measurements.jsonl` beside telemetry when at least one new score is -produced (tool-agnostic artifact). Portable -remote export will reuse the same `OTEL_EXPORTER_OTLP_*` configuration as -agent traces when implemented. +produced (tool-agnostic artifact). When `OTEL_EXPORTER_OTLP_*` is set, those +scores also export as `gen_ai.evaluation.result` span events on the same +TraceID (fail-open; does not rewrite `run-telemetry.jsonl`). Today's scorers (starting with EM-001) read the Level 1/2 **metadata** contract of `run-telemetry.jsonl` — span tree and attributes, not prompt or diff --git a/docs/guides/infrastructure/eval-measurements.md b/docs/guides/infrastructure/eval-measurements.md index 1efd751386..d6103a7cc7 100644 --- a/docs/guides/infrastructure/eval-measurements.md +++ b/docs/guides/infrastructure/eval-measurements.md @@ -28,7 +28,7 @@ computed from that trace (`eval-measurements.jsonl`). The step is Fullsend does not pick an observability product for scores. The portable contract is a local JSONL artifact next to telemetry; remote export reuses the same OpenTelemetry (`OTEL_EXPORTER_OTLP_*`) configuration as agent -traces when implemented. +traces. OTLP (OpenTelemetry Protocol) is the wire format that carries spans and scores to any compatible backend — Phoenix, MLflow, Jaeger, etc. @@ -42,21 +42,25 @@ fullsend run fullsend eval-measure (same GHA job, fail-open, after run) └─ writes output//eval-measurements.jsonl when at least one new score is produced (+ eval-measure-ledger.txt for idempotency) + └─ if OTEL_EXPORTER_OTLP_* set → OTLP export of scores as + gen_ai.evaluation.result span events on the same TraceID + (the W3C Trace ID shared with the agent run — fail-open; + local JSONL always wins) ``` -> **Planned:** portable remote score export via the same `OTEL_EXPORTER_OTLP_*` -> path as agent traces. Not yet implemented. - | Artifact | When | Purpose | |---|---|---| | `run-telemetry.jsonl` | Every run | OTLP JSON TracesData lines (local source of truth for spans) | | `eval-measurements.jsonl` | Every measured run | One JSON object per score (`name`, `label`, `value`, `explanation`, `trace_id`, …). On `label: skip`, `value` is unused (serialized as `0`; ignore it). | | Remote agent spans | OTEL configured | Same spans the local file holds | -| Remote scores *(planned)* | OTEL configured | Scores on the OTLP path — any OTLP backend | +| Remote scores | OTEL configured | Child span `fullsend.eval_measure` + **span event** `gen_ai.evaluation.result` ([normative GenAI events](https://github.com/open-telemetry/semantic-conventions-genai/blob/main/docs/gen-ai/gen-ai-events.md#event-gen_aievaluationresult); [library support matrix](https://github.com/open-telemetry/semantic-conventions-genai/blob/main/reference/reports/gen-ai-evaluation-result-event.md)) correlated by TraceID / parent span ID. Attribute names follow the convention; the convention’s carrier is a **log record** — fullsend uses a span event because only a traces OTLP exporter is configured (log-side consumers will not auto-discover these; a logs-path emit is follow-up). | Orgs choose Phoenix, MLflow, Jaeger, or another collector independently. -Fullsend does not forward vendor-specific score credentials in managed -workflows. +Any OTLP backend can **correlate** scores to the agent run by TraceID. +Vendor score UIs (for example MLflow Assessments panels) may still need a +collector or side consumer that maps the evaluation event — fullsend does +not call those product APIs. Scores are not rewritten into +`run-telemetry.jsonl` (derived products must not mutate primary facts). ## Measurements vs functional evals @@ -255,6 +259,24 @@ least one new measurement row is appended (including `label: skip`). No file is written when telemetry/manifest is missing, no traces match, or every candidate row is already in the ledger. -> **Planned:** portable OTLP score export (same `OTEL_*` as traces) is the -> ADR 0087 remote contract and is not wired yet. Until it lands, consume the -> JSONL artifact (or your own pipeline) for remote dashboards. +When `OTEL_EXPORTER_OTLP_ENDPOINT` or `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` +is set, newly written scores also export as OTLP **span events** +(`fullsend.eval_measure` + `gen_ai.evaluation.result`) on the same +`trace_id`. Attribute names follow the GenAI convention; the convention’s +carrier is a log record — fullsend uses the traces path because only a +traces exporter is configured (see artifact table). Export is fail-open and +does not rewrite `run-telemetry.jsonl`. +The idempotency ledger keys local rows; a remote OTLP failure after a +successful local write will not retry that row on the next run (remote is +best-effort once). Re-export offline by pointing at a fresh out dir, or by +clearing **both** `eval-measure-ledger.txt` and `eval-measurements.jsonl` — +clearing only the ledger re-appends duplicate rows to the JSONL +(`AppendMeasurements` is `O_APPEND` with no dedup). + +Managed measure assumes one platform `run-telemetry.jsonl` per runDir (each +`fullsend run` creates a unique `output/fs--/`). If inbound +`TRACEPARENT` is present and unsampled, score export skips only rows whose +`trace_id` matches that parent TraceID (same orphan-avoidance rule as agent +`parentSampledProcessor`); other TraceIDs in the batch still export. +Cross-run cost/correlation rollup is out of scope here (see hierarchical +work-graph IDs). diff --git a/docs/problems/operational-observability.md b/docs/problems/operational-observability.md index 92a491ff6e..5f8c2b9049 100644 --- a/docs/problems/operational-observability.md +++ b/docs/problems/operational-observability.md @@ -192,7 +192,7 @@ This works for early experimentation when the volume is low and the operators ar - How should trace access be controlled? (JSONL trace exposure decided in [ADR 0021](../ADRs/0021-jsonl-reasoning-trace-exposure.md): owner-scoped storage with credential scanning as defense-in-depth. Broader question of balancing security and transparency for non-JSONL observability data remains open.) - What retention policy applies to traces? Indefinite retention supports audit requirements but increases storage cost and data sensitivity exposure. Time-bounded retention (e.g., 90 days) limits exposure but may lose traces needed for incident investigation. - How do we measure "is the system getting better"? What metrics constitute a meaningful quality signal for an autonomous software factory? Merge revert rate? Human override rate? Time-to-review? Cost per decision? Some composite score? The choice of metric shapes what gets optimized. First-ship trend scores (trace fitness on wild runs, local `eval-measurements.jsonl`) are [ADR 0087](../ADRs/0087-eval-measurements-online-trace-scoring.md); richer quality signals remain open. -- At what scale does a dedicated LLM observability platform justify its operational overhead (Postgres, ClickHouse, Redis, S3 for something like Langfuse)? Is there a threshold of agent activity below which structured logging suffices? Score files are local JSONL ([ADR 0087](../ADRs/0087-eval-measurements-online-trace-scoring.md)); remote scores reuse `OTEL_EXPORTER_OTLP_*` when implemented. Platform choice remains open. +- At what scale does a dedicated LLM observability platform justify its operational overhead (Postgres, ClickHouse, Redis, S3 for something like Langfuse)? Is there a threshold of agent activity below which structured logging suffices? Score files are local JSONL ([ADR 0087](../ADRs/0087-eval-measurements-online-trace-scoring.md)); remote scores reuse `OTEL_EXPORTER_OTLP_*`. Platform choice remains open. - ~~How do we handle the bootstrapping problem — the factory needs observability to improve, but building the observability infrastructure is itself work that competes with building the factory?~~ Decided in [ADR 0050](../ADRs/0050-distributed-tracing-instrumentation.md): zero-configuration baseline (local JSONL + summary files) eliminates infrastructure requirements for initial observability; OTLP export adds backends when the org is ready. - Should observability data feed back into agent instructions automatically (e.g., auto-adjusting prompts when false positive rates exceed a threshold), or should it only inform human-driven instruction changes? Automatic feedback creates the risk of instruction oscillation; human-only feedback is slower but more controlled. - How do we build community dashboards that are useful to contributors with different levels of technical depth — from "is the agent doing a good job on my repo" to "show me the trace of this specific review"? diff --git a/hack/prove-otlp-scores/main.go b/hack/prove-otlp-scores/main.go new file mode 100644 index 0000000000..ab31567e00 --- /dev/null +++ b/hack/prove-otlp-scores/main.go @@ -0,0 +1,192 @@ +// Command prove-otlp-scores scores a real run-telemetry.jsonl and asserts +// portable OTLP gen_ai.evaluation.result events arrive at a local sink. +package main + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" + "google.golang.org/protobuf/proto" + + "github.com/fullsend-ai/fullsend/internal/evalmeasure" +) + +func main() { + if len(os.Args) < 3 { + fmt.Fprintf(os.Stderr, "usage: %s [out-dir]\n", os.Args[0]) + fmt.Fprintf(os.Stderr, " out-dir defaults to a fresh temp dir (never the telemetry file's directory).\n") + os.Exit(2) + } + telem := os.Args[1] + reg := os.Args[2] + out := "" + if len(os.Args) > 3 { + out = os.Args[3] + } else { + tmp, err := os.MkdirTemp("", "prove-otlp-scores-*") + if err != nil { + fmt.Fprintf(os.Stderr, "temp out-dir: %v\n", err) + os.Exit(1) + } + out = tmp + fmt.Fprintf(os.Stderr, "using temp out-dir %s\n", out) + } + if err := os.MkdirAll(out, 0o755); err != nil { + fmt.Fprintf(os.Stderr, "out-dir: %v\n", err) + os.Exit(1) + } + // Only wipe ledger/measurements inside the chosen out-dir (temp by + // default), never beside a live run-telemetry.jsonl by accident. + _ = os.Remove(filepath.Join(out, evalmeasure.LedgerFile)) + _ = os.Remove(filepath.Join(out, evalmeasure.MeasurementsFile)) + + var mu sync.Mutex + var reqs []*coltracepb.ExportTraceServiceRequest + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if r.Header.Get("Content-Encoding") == "gzip" { + zr, err := gzip.NewReader(bytes.NewReader(raw)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + raw, err = io.ReadAll(zr) + _ = zr.Close() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + } + var req coltracepb.ExportTraceServiceRequest + if err := proto.Unmarshal(raw, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + mu.Lock() + reqs = append(reqs, &req) + mu.Unlock() + resp, _ := proto.Marshal(&coltracepb.ExportTraceServiceResponse{}) + w.Header().Set("Content-Type", "application/x-protobuf") + _, _ = w.Write(resp) + })) + defer srv.Close() + + _ = os.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", srv.URL) + _ = os.Unsetenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") + _ = os.Unsetenv("OTEL_SDK_DISABLED") + // Clear ambient W3C parents so an unsampled TRACEPARENT from a prior + // fullsend run in this shell cannot suppress every score and make the + // prove tool report a false FAIL. + _ = os.Unsetenv("TRACEPARENT") + _ = os.Unsetenv("TRACESTATE") + + results, stats, err := evalmeasure.MeasureAndExport(context.Background(), telem, reg, out, "dev") + if err != nil { + fmt.Fprintf(os.Stderr, "measure failed: %v\n", err) + os.Exit(1) + } + + mu.Lock() + reqsCopy := append([]*coltracepb.ExportTraceServiceRequest(nil), reqs...) + nReqs := len(reqs) + mu.Unlock() + + events := extractEvents(reqsCopy) + report := map[string]any{ + "endpoint": srv.URL, + "out_dir": out, + "scores_written": len(results), + "remote_export_warning": stats.RemoteExportWarning, + "results": results, + "otlp_requests": nReqs, + "events": events, + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + _ = enc.Encode(report) + + if len(results) == 0 { + fmt.Fprintf(os.Stderr, "FAIL: no scores written\n") + os.Exit(1) + } + if nReqs == 0 { + fmt.Fprintf(os.Stderr, "FAIL: no OTLP requests received\n") + os.Exit(1) + } + if len(events) == 0 { + fmt.Fprintf(os.Stderr, "FAIL: no gen_ai.evaluation.result events\n") + os.Exit(1) + } + fmt.Fprintf(os.Stderr, "PASS: %d score(s), %d OTLP event(s)\n", len(results), len(events)) +} + +type eventView struct { + SpanName string `json:"span_name"` + TraceID string `json:"trace_id"` + ParentID string `json:"parent_span_id"` + EventName string `json:"event_name"` + Attributes map[string]any `json:"attributes"` +} + +func extractEvents(reqs []*coltracepb.ExportTraceServiceRequest) []eventView { + var out []eventView + for _, req := range reqs { + for _, rs := range req.GetResourceSpans() { + for _, ss := range rs.GetScopeSpans() { + for _, sp := range ss.GetSpans() { + for _, ev := range sp.GetEvents() { + if ev.GetName() != evalmeasure.EventGenAIEvaluationResult { + continue + } + attrs := map[string]any{} + for _, kv := range ev.GetAttributes() { + attrs[kv.GetKey()] = anyValue(kv.GetValue()) + } + out = append(out, eventView{ + SpanName: sp.GetName(), + TraceID: hex.EncodeToString(sp.GetTraceId()), + ParentID: hex.EncodeToString(sp.GetParentSpanId()), + EventName: ev.GetName(), + Attributes: attrs, + }) + } + } + } + } + } + return out +} + +func anyValue(v *commonpb.AnyValue) any { + if v == nil { + return nil + } + switch x := v.GetValue().(type) { + case *commonpb.AnyValue_StringValue: + return x.StringValue + case *commonpb.AnyValue_DoubleValue: + return x.DoubleValue + case *commonpb.AnyValue_IntValue: + return x.IntValue + case *commonpb.AnyValue_BoolValue: + return x.BoolValue + default: + return v.String() + } +} diff --git a/internal/cli/evalmeasure.go b/internal/cli/evalmeasure.go index e3db77ed4f..f769600f38 100644 --- a/internal/cli/evalmeasure.go +++ b/internal/cli/evalmeasure.go @@ -35,14 +35,19 @@ func newEvalMeasureCmd() *cobra.Command { Long: `Parse run-telemetry.jsonl, score with an agents measurement manifest, and write eval-measurements.jsonl beside the telemetry artifact. +When OTEL_EXPORTER_OTLP_ENDPOINT or OTEL_EXPORTER_OTLP_TRACES_ENDPOINT is +set (same env as agent traces, ADR 0050), newly written scores are also +exported as OTLP span events (gen_ai.evaluation.result) on the scored +trace. Export is fail-open: local JSONL always wins. + The binary resolves the manifest (local FULLSEND_DIR override, else a SHA-pinned fetch from fullsend-ai/agents — same pin, allowlist, hash, and audit as harness fallback). Platform telemetry is the file at the top of each run directory; nested iteration-N/output/ copies are ignored. Remote backends are not selected by fullsend: scores are a portable local -JSONL artifact. When portable OTLP score export lands, it will reuse the -same OTEL_EXPORTER_OTLP_* configuration as agent traces (ADR 0050 / 0087). +JSONL artifact plus optional OTLP on the shared OTEL_* path (ADR 0087). +No vendor score adapters (MLflow Assessments, Phoenix SDK, …) in core. Exit 0 when scores fail — measurements are data, not gates. Non-zero only on hard IO/parse errors. Missing telemetry or manifest is a skip (exit 0).`, @@ -112,7 +117,7 @@ func runEvalMeasure(ctx context.Context, printer *ui.Printer, opts evalMeasureOp var all []evalmeasure.EvaluationResult for _, p := range telemPaths { - results, stats, err := evalmeasure.MeasureAndExport(ctx, p, registry, opts.outDir) + results, stats, err := evalmeasure.MeasureAndExport(ctx, p, registry, opts.outDir, Version()) if stats.Incomplete != "" { printer.StepWarn(fmt.Sprintf("%s: telemetry parse incomplete (%s); scored available traces", p, stats.Incomplete)) } @@ -122,6 +127,9 @@ func runEvalMeasure(ctx context.Context, printer *ui.Printer, opts evalMeasureOp if stats.SkippedSpans > 0 { printer.StepWarn(fmt.Sprintf("%s: skipped %d unreadable span(s) inside otherwise-valid telemetry line(s)", p, stats.SkippedSpans)) } + if stats.RemoteExportWarning != "" { + printer.StepWarn(fmt.Sprintf("%s: OTLP score export failed (local JSONL kept): %s", p, stats.RemoteExportWarning)) + } if err != nil { return append(all, results...), false, err } diff --git a/internal/cli/evalmeasure_test.go b/internal/cli/evalmeasure_test.go index fad628a186..50eed1f9d5 100644 --- a/internal/cli/evalmeasure_test.go +++ b/internal/cli/evalmeasure_test.go @@ -17,7 +17,20 @@ import ( "github.com/fullsend-ai/fullsend/internal/ui" ) +// clearOTLPEnv keeps Measure*/eval-measure tests hermetic when CI injects +// OTEL_EXPORTER_OTLP_* org vars (same pattern as evalmeasure/export_otlp_test). +func clearOTLPEnv(t *testing.T) { + t.Helper() + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "") + t.Setenv("OTEL_EXPORTER_OTLP_HEADERS", "") + t.Setenv("OTEL_SDK_DISABLED", "") + t.Setenv("TRACEPARENT", "") +} + func TestEvalMeasureCmd_ScoresFixture(t *testing.T) { + clearOTLPEnv(t) out := t.TempDir() telemetryPath := filepath.Join("..", "evalmeasure", "testdata", "complete.jsonl") registry := filepath.Join("..", "evalmeasure", "testdata", "sample-registry.yaml") @@ -71,6 +84,7 @@ func TestEvalMeasureCmd_MissingRequiredFlags(t *testing.T) { } func TestEvalMeasureCmd_OutputDirIgnoresNestedTelemetry_LegacyFormat(t *testing.T) { + clearOTLPEnv(t) fsDir := t.TempDir() outBase := t.TempDir() runDir := filepath.Join(outBase, "agent-triage-1-1") @@ -113,6 +127,7 @@ func TestEvalMeasureCmd_OutputDirIgnoresNestedTelemetry_LegacyFormat(t *testing. // YAML: a local FULLSEND_DIR eval/measurements/.yaml must produce // eval-measurements.jsonl. func TestEvalMeasureCmd_LocalFullsendDirManifestProducesJSONL_LegacyFormat(t *testing.T) { + clearOTLPEnv(t) fsDir := t.TempDir() outBase := t.TempDir() runDir := filepath.Join(outBase, "agent-triage-2-2") @@ -147,6 +162,7 @@ func TestEvalMeasureCmd_LocalFullsendDirManifestProducesJSONL_LegacyFormat(t *te } func TestEvalMeasureCmd_OutputDirIgnoresNestedTelemetry_NewFormat(t *testing.T) { + clearOTLPEnv(t) fsDir := t.TempDir() outBase := t.TempDir() runDir := filepath.Join(outBase, "fs-tri-aabbccddee00") @@ -185,6 +201,7 @@ func TestEvalMeasureCmd_OutputDirIgnoresNestedTelemetry_NewFormat(t *testing.T) } func TestEvalMeasureCmd_LocalFullsendDirManifestProducesJSONL_NewFormat(t *testing.T) { + clearOTLPEnv(t) fsDir := t.TempDir() outBase := t.TempDir() runDir := filepath.Join(outBase, "fs-tri-1122334455ff") @@ -230,6 +247,7 @@ func writeTwoTraceTelemetry(t *testing.T) string { } func TestRunEvalMeasure_ErrorIncludesPartialResults(t *testing.T) { + clearOTLPEnv(t) out := t.TempDir() telem := writeTwoTraceTelemetry(t) registry := filepath.Join("..", "evalmeasure", "testdata", "sample-registry.yaml") @@ -255,6 +273,7 @@ func TestRunEvalMeasure_ErrorIncludesPartialResults(t *testing.T) { } func TestEvalMeasureCmd_ErrorPrintsPartialFromFailingFile(t *testing.T) { + clearOTLPEnv(t) out := t.TempDir() telem := writeTwoTraceTelemetry(t) ctx := evalmeasure.WithPersistHook(context.Background(), func() { @@ -281,6 +300,7 @@ func TestEvalMeasureCmd_ErrorPrintsPartialFromFailingFile(t *testing.T) { } func TestEvalMeasureCmd_ErrorDoesNotPrintWrote(t *testing.T) { + clearOTLPEnv(t) out := filepath.Join(t.TempDir(), "not-a-dir") require.NoError(t, os.WriteFile(out, []byte("x"), 0o644)) @@ -300,6 +320,7 @@ func TestEvalMeasureCmd_ErrorDoesNotPrintWrote(t *testing.T) { } func TestEvalMeasureCmd_SkipWhenNoTelemetry(t *testing.T) { + clearOTLPEnv(t) cmd := newRootCmd() buf := &bytes.Buffer{} cmd.SetOut(buf) @@ -315,6 +336,7 @@ func TestEvalMeasureCmd_SkipWhenNoTelemetry(t *testing.T) { } func TestEvalMeasureCmd_WarnsOnCorruptTelemetryLine(t *testing.T) { + clearOTLPEnv(t) out := t.TempDir() good, err := os.ReadFile(filepath.Join("..", "evalmeasure", "testdata", "complete.jsonl")) require.NoError(t, err) @@ -336,6 +358,7 @@ func TestEvalMeasureCmd_WarnsOnCorruptTelemetryLine(t *testing.T) { } func TestEvalMeasureCmd_WarnsOnIncompleteParse(t *testing.T) { + clearOTLPEnv(t) out := t.TempDir() good, err := os.ReadFile(filepath.Join("..", "evalmeasure", "testdata", "complete.jsonl")) require.NoError(t, err) @@ -483,6 +506,7 @@ func TestResolveEvalMeasureRegistry_RejectsPathAgent(t *testing.T) { } func TestEvalMeasureCmd_TelemetryWithoutRegistry(t *testing.T) { + clearOTLPEnv(t) cmd := newRootCmd() buf := &bytes.Buffer{} cmd.SetOut(buf) @@ -515,3 +539,25 @@ func TestPrintMeasurementResults_SkipAndNoWroteOnError(t *testing.T) { }}, false) assert.NotContains(t, buf.String(), "Wrote") } + +func TestRunEvalMeasure_OTLPFailWarns(t *testing.T) { + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:1") + out := t.TempDir() + telemetryPath := filepath.Join("..", "evalmeasure", "testdata", "complete.jsonl") + registry := filepath.Join("..", "evalmeasure", "testdata", "sample-registry.yaml") + var buf bytes.Buffer + results, skipped, err := runEvalMeasure(context.Background(), ui.New(&buf), evalMeasureOpts{ + telemetryPath: telemetryPath, + registryPath: registry, + outDir: out, + }) + require.NoError(t, err) + assert.False(t, skipped) + require.NotEmpty(t, results) + assert.Contains(t, buf.String(), "OTLP score export failed") + assert.Contains(t, buf.String(), "otlp export failed for all", + "transport failure must not claim N/M row-construction success") + _, statErr := os.Stat(filepath.Join(out, evalmeasure.MeasurementsFile)) + require.NoError(t, statErr, "local JSONL must still be written") +} diff --git a/internal/evalmeasure/export_otlp.go b/internal/evalmeasure/export_otlp.go new file mode 100644 index 0000000000..d51261b8d2 --- /dev/null +++ b/internal/evalmeasure/export_otlp.go @@ -0,0 +1,342 @@ +package evalmeasure + +import ( + "context" + "encoding/hex" + "errors" + "fmt" + "os" + "strings" + "sync" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/propagation" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/trace" + + "github.com/fullsend-ai/fullsend/internal/telemetry" +) + +// GenAI evaluation event / attribute names. +// +// Attribute names follow OpenTelemetry GenAI semantic conventions +// (normative pin consulted for this ship: +// https://github.com/open-telemetry/semantic-conventions-genai/blob/main/docs/gen-ai/gen-ai-events.md#event-gen_aievaluationresult +// — GenAI events live in semantic-conventions-genai; treat as low-stability. +// Supporting-library matrix only: +// https://github.com/open-telemetry/semantic-conventions-genai/blob/main/reference/reports/gen-ai-evaluation-result-event.md). +// Semconv remains unstable across minor versions (see gen_ai.system → +// gen_ai.provider.name); bump measurement versions when attribute names change. +// +// Carrier deviation (deliberate): the convention defines +// gen_ai.evaluation.result as a log-record / Event-API event (SHOULD parent +// to the GenAI operation span, or set gen_ai.response.id when span id is +// unavailable). Fullsend emits it as a span event via span.AddEvent on a +// short fullsend.eval_measure child over the traces OTLP path, because only +// a traces exporter is configured today — log-side consumers will not +// auto-discover these scores. A conforming log-record emit is follow-up +// work when a logs exporter exists; this PR does not add one. +// +// Post-hoc attach: the scored GenAI operation span is already flushed, so +// the child span is remote-parented to that SpanID. Any OTLP traces backend +// can correlate by TraceID; vendor score UIs (Assessments panels, etc.) may +// still need a collector/consumer mapping — fullsend does not call those +// APIs. +const ( + EventGenAIEvaluationResult = "gen_ai.evaluation.result" + + AttrGenAIEvaluationName = "gen_ai.evaluation.name" + AttrGenAIEvaluationScoreValue = "gen_ai.evaluation.score.value" + AttrGenAIEvaluationScoreLabel = "gen_ai.evaluation.score.label" + AttrGenAIEvaluationExplanation = "gen_ai.evaluation.explanation" + AttrFullsendMeasurementVersion = "fullsend.measurement.version" + + spanNameEvalMeasure = "fullsend.eval_measure" + otlpScopeName = "github.com/fullsend-ai/fullsend/internal/evalmeasure" + // Score export is post-hoc and fail-open: bound total wall time for + // exporter create + span emit + ForceFlush + Shutdown so a flaky + // collector cannot hang the agent job until GHA timeout. + otlpExportBudget = 15 * time.Second + otlpRetryBudget = 5 * time.Second +) + +// newScoreOTLPExporter is a test seam. Production uses a retry-bounded +// exporter so Simple/Batch export cannot retry forever. Live agent Setup's +// NewOTLPExporter sets InitialInterval/MaxInterval but leaves +// MaxElapsedTime at 0, which the otlptracehttp retry loop treats as "never +// give up on elapsed time" (bounded only by shutdown ctx cancellation) — +// scores intentionally stay tighter. +var newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporterBounded(ctx, otlpRetryBudget) +} + +// ExportOTLPScores emits each measurement as a short child span on the same +// TraceID (remote-parented to the scored SpanID) with a +// gen_ai.evaluation.result event. Uses the same OTEL_EXPORTER_OTLP_* env as +// ADR 0050 agent traces. serviceVersion must match telemetry.Setup's version +// (CLI Version()) so resource identity stays aligned with agent spans. +// No-op when OTEL is unset or OTEL_SDK_DISABLED=true. Per-score: when inbound +// TRACEPARENT is valid and unsampled, scores whose TraceID equals that +// parent TraceID are skipped (same rule as parentSampledProcessor — avoid +// orphaning score spans on a TraceID that never left the box). Other +// TraceIDs in the batch are unaffected. Fail-open: returns an error for the +// caller to warn on; never writes primary telemetry files. Rows with +// empty/zero IDs are skipped. Pure ID failures report "N/M scores +// exported" so operators can tell partial from total failure; a +// ForceFlush/export transport error uses a distinct "failed for all N" +// message (row construction is not treated as delivery). +func ExportOTLPScores(ctx context.Context, results []EvaluationResult, serviceVersion string) (err error) { + if len(results) == 0 { + return nil + } + if sdkDisable := os.Getenv("OTEL_SDK_DISABLED"); strings.EqualFold(strings.TrimSpace(sdkDisable), "true") { + return nil + } + if !telemetry.OTLPEnabled() { + return nil + } + if err := telemetry.ValidateOTLPEndpoints(); err != nil { + return fmt.Errorf("otlp endpoint validation: %w", err) + } + + ctx, cancel := context.WithTimeout(ctx, otlpExportBudget) + defer cancel() + + exp, err := newScoreOTLPExporter(ctx) + if err != nil { + return fmt.Errorf("otlp exporter: %w", err) + } + capExp := &capturingExporter{base: exp} + + // Batch so N scores share one (or few) HTTP exports under the budget. + tp := sdktrace.NewTracerProvider( + sdktrace.WithResource(telemetry.BuildResource(serviceVersion)), + sdktrace.WithSampler(sdktrace.AlwaysSample()), + sdktrace.WithRawSpanLimits(telemetry.SpanLimits()), + sdktrace.WithSpanProcessor(sdktrace.NewBatchSpanProcessor(capExp)), + ) + // Shutdown shares the same export budget (remaining deadline on ctx), + // not an extra Background timeout stacked on top. + defer func() { + if shutErr := tp.Shutdown(ctx); shutErr != nil { + err = errors.Join(err, fmt.Errorf("otlp shutdown: %w", shutErr)) + } + }() + + suppressTID, suppressUnsampled := inboundUnsampledTRACEPARENT() + tr := tp.Tracer(otlpScopeName) + var ( + errs []error + attempted int + failed int + ) + for _, r := range results { + if suppressUnsampled && scoreTraceIDEquals(r.TraceID, suppressTID) { + // Inbound parent for this TraceID was unsampled — agent OTLP + // suppressed; do not orphan a score span on that TraceID. + continue + } + if strings.TrimSpace(r.TraceID) == "" || strings.TrimSpace(r.SpanID) == "" { + // Same silent skip as exportOneScore (no parent to correlate). + continue + } + attempted++ + if err := exportOneScore(ctx, tr, r); err != nil { + failed++ + errs = append(errs, err) + } + } + var flushErr error + if flushErr = tp.ForceFlush(ctx); flushErr != nil { + errs = append(errs, flushErr) + } + capExp.mu.Lock() + expErr := capExp.err + capExp.mu.Unlock() + if expErr != nil { + errs = append(errs, fmt.Errorf("otlp export: %w", expErr)) + } + if len(errs) == 0 { + return nil + } + // Transport failure: nothing is known to have landed — do not claim N/M + // success from spans that were only constructed locally. + if flushErr != nil || expErr != nil { + return fmt.Errorf("otlp export failed for all %d scores: %w", attempted, errors.Join(errs...)) + } + exported := attempted - failed + return fmt.Errorf("%d/%d scores exported; %d failed: %w", exported, attempted, failed, errors.Join(errs...)) +} + +// inboundUnsampledTRACEPARENT parses TRACEPARENT with the same W3C +// TraceContext propagator as fullsend run. When the inbound parent is valid, +// remote, and unsampled, it returns that TraceID and true. Empty/malformed +// or sampled TRACEPARENT → false (export proceeds for all scores). +func inboundUnsampledTRACEPARENT() (trace.TraceID, bool) { + tp := strings.TrimSpace(os.Getenv("TRACEPARENT")) + if tp == "" { + return trace.TraceID{}, false + } + ctx := propagation.TraceContext{}.Extract(context.Background(), propagation.MapCarrier{ + "traceparent": tp, + "tracestate": strings.TrimSpace(os.Getenv("TRACESTATE")), + }) + sc := trace.SpanContextFromContext(ctx) + if !sc.IsValid() || !sc.IsRemote() || sc.IsSampled() { + return trace.TraceID{}, false + } + return sc.TraceID(), true +} + +func scoreTraceIDEquals(hexID string, want trace.TraceID) bool { + got, err := parseTraceID(hexID) + if err != nil { + return false + } + return got == want +} + +// capturingExporter tracks the latest ExportSpans error so fail-open callers +// can warn. A later successful ExportSpans clears a prior transient failure +// (avoids false "remote export failed" after data actually landed). Mutex +// covers BatchSpanProcessor's async export goroutine. +type capturingExporter struct { + base sdktrace.SpanExporter + mu sync.Mutex + err error +} + +func (c *capturingExporter) ExportSpans(ctx context.Context, spans []sdktrace.ReadOnlySpan) error { + err := c.base.ExportSpans(ctx, spans) + c.mu.Lock() + c.err = err + c.mu.Unlock() + return err +} + +func (c *capturingExporter) Shutdown(ctx context.Context) error { + return c.base.Shutdown(ctx) +} + +func exportOneScore(ctx context.Context, tr trace.Tracer, r EvaluationResult) error { + if strings.TrimSpace(r.TraceID) == "" || strings.TrimSpace(r.SpanID) == "" { + // EM-001 skip rows can omit span_id when the root run span is missing. + // Local JSONL still records them; OTLP needs a parent to correlate. + return nil + } + tid, err := parseTraceID(r.TraceID) + if err != nil { + return fmt.Errorf("trace_id %q: %w", r.TraceID, err) + } + sid, err := parseSpanID(r.SpanID) + if err != nil { + return fmt.Errorf("span_id %q: %w", r.SpanID, err) + } + + psc := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: tid, + SpanID: sid, + TraceFlags: trace.FlagsSampled, + Remote: true, + }) + parent := trace.ContextWithRemoteSpanContext(ctx, psc) + + attrs := []attribute.KeyValue{ + attribute.String(AttrGenAIAgentName, r.Agent), + attribute.String(AttrFullsendMeasurementVersion, r.Version), + } + if r.WorkItemID != "" { + attrs = append(attrs, attribute.String(AttrFullsendWorkItemID, r.WorkItemID)) + } + + _, span := tr.Start(parent, spanNameEvalMeasure, trace.WithAttributes(attrs...)) + defer span.End() + + eventAttrs := []attribute.KeyValue{ + attribute.String(AttrGenAIEvaluationName, r.Name), + attribute.String(AttrGenAIEvaluationScoreLabel, r.Label), + // Event attribute values are not truncated by SpanLimits (SDK + // applies AttributeValueLengthLimit to span attrs only). Bound + // explanation at the call site via FreeTextAttrValueLenLimit — + // operator OTEL_* overrides (incl. -1) or MaxSpanAttrValueLen — + // not content-capture-aware SpanLimits (Level 3 would leave it + // unbounded). + attribute.String(AttrGenAIEvaluationExplanation, truncateRunes(r.Explanation, eventAttrValueLenLimit())), + attribute.String(AttrFullsendMeasurementVersion, r.Version), + } + // Skip rows leave Value unused (serialized as 0 in JSONL); do not publish + // a numeric zero that backends may chart as a real score. + if r.Label != LabelSkip { + eventAttrs = append(eventAttrs, attribute.Float64(AttrGenAIEvaluationScoreValue, r.Value)) + } + + span.AddEvent(EventGenAIEvaluationResult, trace.WithAttributes(eventAttrs...)) + // Keep Ok for all labels: pass/fail/skip live on the evaluation event. + // Error status would conflate derived fitness fail with run failure in + // backends that key off span status. + span.SetStatus(codes.Ok, "") + return nil +} + +func truncateRunes(s string, max int) string { + if max < 0 { + return s + } + n := 0 + for i := range s { + if n == max { + return s[:i] + } + n++ + } + return s +} + +// eventAttrValueLenLimit bounds gen_ai.evaluation.explanation for OTLP. +// Uses telemetry.FreeTextAttrValueLenLimit: operator OTEL_* limit when set +// (including explicit -1 = unlimited), else MaxSpanAttrValueLen. Does not +// reuse SpanLimits(), whose negative sentinel under Level 3 content capture +// would leave explanations unbounded on the wire. +func eventAttrValueLenLimit() int { + return telemetry.FreeTextAttrValueLenLimit() +} + +func parseTraceID(hexID string) (trace.TraceID, error) { + var out trace.TraceID + b, err := decodeFixedHex(hexID, len(out)) + if err != nil { + return out, err + } + copy(out[:], b) + if !out.IsValid() { + return out, fmt.Errorf("trace_id must be non-zero") + } + return out, nil +} + +func parseSpanID(hexID string) (trace.SpanID, error) { + var out trace.SpanID + b, err := decodeFixedHex(hexID, len(out)) + if err != nil { + return out, err + } + copy(out[:], b) + if !out.IsValid() { + return out, fmt.Errorf("span_id must be non-zero") + } + return out, nil +} + +func decodeFixedHex(s string, n int) ([]byte, error) { + s = strings.TrimSpace(s) + if len(s) != n*2 { + return nil, fmt.Errorf("want %d hex chars, got %d", n*2, len(s)) + } + b, err := hex.DecodeString(s) + if err != nil { + return nil, err + } + return b, nil +} diff --git a/internal/evalmeasure/export_otlp_test.go b/internal/evalmeasure/export_otlp_test.go new file mode 100644 index 0000000000..6f60debdd6 --- /dev/null +++ b/internal/evalmeasure/export_otlp_test.go @@ -0,0 +1,491 @@ +package evalmeasure + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/hex" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + "google.golang.org/protobuf/proto" + + "github.com/fullsend-ai/fullsend/internal/telemetry" +) + +type scoreOTLPSink struct { + mu sync.Mutex + reqs []*coltracepb.ExportTraceServiceRequest + srv *httptest.Server +} + +func newScoreOTLPSink(t *testing.T) *scoreOTLPSink { + t.Helper() + s := &scoreOTLPSink{} + s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if r.Header.Get("Content-Encoding") == "gzip" { + zr, err := gzip.NewReader(bytes.NewReader(raw)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + raw, err = io.ReadAll(zr) + _ = zr.Close() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + } + var req coltracepb.ExportTraceServiceRequest + if err := proto.Unmarshal(raw, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + s.mu.Lock() + s.reqs = append(s.reqs, &req) + s.mu.Unlock() + resp, _ := proto.Marshal(&coltracepb.ExportTraceServiceResponse{}) + w.Header().Set("Content-Type", "application/x-protobuf") + _, _ = w.Write(resp) + })) + t.Cleanup(s.srv.Close) + return s +} + +func (s *scoreOTLPSink) allSpans() []*coltracepb.ExportTraceServiceRequest { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]*coltracepb.ExportTraceServiceRequest, len(s.reqs)) + copy(out, s.reqs) + return out +} + +func clearOTLPEnv(t *testing.T) { + t.Helper() + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "") + t.Setenv("OTEL_EXPORTER_OTLP_HEADERS", "") + t.Setenv("OTEL_SDK_DISABLED", "") + t.Setenv("TRACEPARENT", "") + t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "") + t.Setenv("OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT", "") + t.Setenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "") +} + +func TestExportOTLPScores_NoopWithoutEndpoint(t *testing.T) { + clearOTLPEnv(t) + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", TraceID: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", SpanID: "bbbbbbbbbbbbbbbb", + }}, "test-1.2.3") + require.NoError(t, err) +} + +func TestExportOTLPScores_EmitsGenAIEvaluationEvent(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL+"/v1/traces") + + orig := newScoreOTLPExporter + t.Cleanup(func() { newScoreOTLPExporter = orig }) + newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporter(ctx) + } + + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", + Label: LabelPass, + Explanation: "span_tree=pass", + TraceID: "84d470ba2451ffeccfe09022d9b2aebd", + SpanID: "77f8c0902eaeedcb", + WorkItemID: "fullsend-ai/fullsend#6449", + Agent: "review", + Version: "em-001@1", + Value: 1.0, + }}, "test-1.2.3") + require.NoError(t, err) + + reqs := sink.allSpans() + require.NotEmpty(t, reqs) + + var foundEvent bool + var spanName string + var parentHex string + var traceHex string + var svcName, svcVer string + for _, req := range reqs { + for _, rs := range req.GetResourceSpans() { + for _, kv := range rs.GetResource().GetAttributes() { + switch kv.GetKey() { + case "service.name": + svcName = kv.GetValue().GetStringValue() + case "service.version": + svcVer = kv.GetValue().GetStringValue() + } + } + for _, ss := range rs.GetScopeSpans() { + for _, sp := range ss.GetSpans() { + spanName = sp.GetName() + traceHex = hex.EncodeToString(sp.GetTraceId()) + parentHex = hex.EncodeToString(sp.GetParentSpanId()) + for _, ev := range sp.GetEvents() { + if ev.GetName() != EventGenAIEvaluationResult { + continue + } + foundEvent = true + attrs := map[string]string{} + var score float64 + var hasScore bool + for _, kv := range ev.GetAttributes() { + switch kv.GetKey() { + case AttrGenAIEvaluationName, AttrGenAIEvaluationScoreLabel, AttrGenAIEvaluationExplanation, AttrFullsendMeasurementVersion: + attrs[kv.GetKey()] = kv.GetValue().GetStringValue() + case AttrGenAIEvaluationScoreValue: + score = kv.GetValue().GetDoubleValue() + hasScore = true + } + } + assert.Equal(t, "trace_fitness", attrs[AttrGenAIEvaluationName]) + assert.Equal(t, LabelPass, attrs[AttrGenAIEvaluationScoreLabel]) + assert.Equal(t, "span_tree=pass", attrs[AttrGenAIEvaluationExplanation]) + assert.Equal(t, "em-001@1", attrs[AttrFullsendMeasurementVersion]) + require.True(t, hasScore) + assert.Equal(t, 1.0, score) + } + } + } + } + } + require.True(t, foundEvent, "expected gen_ai.evaluation.result event") + assert.Equal(t, spanNameEvalMeasure, spanName) + assert.Equal(t, "84d470ba2451ffeccfe09022d9b2aebd", traceHex) + assert.Equal(t, "77f8c0902eaeedcb", parentHex) + assert.Equal(t, "fullsend", svcName) + assert.Equal(t, "test-1.2.3", svcVer) +} + +func TestExportOTLPScores_InvalidTraceID(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", TraceID: "not-a-trace", SpanID: "77f8c0902eaeedcb", + }}, "test-1.2.3") + require.Error(t, err) +} + +func TestExportOTLPScores_InvalidSpanIDHex(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", + TraceID: "84d470ba2451ffeccfe09022d9b2aebd", + SpanID: "zzzzzzzzzzzzzzzz", + }}, "test-1.2.3") + require.Error(t, err) +} + +func TestExportOTLPScores_DisabledSDK(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_SDK_DISABLED", "true") + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", TraceID: "84d470ba2451ffeccfe09022d9b2aebd", SpanID: "77f8c0902eaeedcb", + }}, "test-1.2.3") + require.NoError(t, err) + assert.Empty(t, sink.allSpans()) +} + +func TestExportOTLPScores_EmptySpanIDSkipped(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", Label: LabelSkip, TraceID: "84d470ba2451ffeccfe09022d9b2aebd", SpanID: "", + }}, "test-1.2.3") + require.NoError(t, err) + assert.Empty(t, sink.allSpans()) +} + +func TestExportOTLPScores_ZeroTraceID(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", TraceID: "00000000000000000000000000000000", SpanID: "77f8c0902eaeedcb", + }}, "test-1.2.3") + require.Error(t, err) +} + +func TestExportOTLPScores_SkipOmitsScoreValue(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", Label: LabelSkip, Explanation: "no run span", + TraceID: "84d470ba2451ffeccfe09022d9b2aebd", SpanID: "77f8c0902eaeedcb", + Agent: "review", Version: "em-001@1", Value: 0, + }}, "test-1.2.3") + require.NoError(t, err) + reqs := sink.allSpans() + require.NotEmpty(t, reqs) + for _, req := range reqs { + for _, rs := range req.GetResourceSpans() { + for _, ss := range rs.GetScopeSpans() { + for _, sp := range ss.GetSpans() { + for _, ev := range sp.GetEvents() { + if ev.GetName() != EventGenAIEvaluationResult { + continue + } + for _, kv := range ev.GetAttributes() { + assert.NotEqual(t, AttrGenAIEvaluationScoreValue, kv.GetKey(), "skip must omit score.value") + } + } + } + } + } + } +} + +func TestMeasureAndExport_OTLPFailOpen(t *testing.T) { + dir := t.TempDir() + telemSrc := filepath.Join("testdata", "complete.jsonl") + telem := filepath.Join(dir, "run-telemetry.jsonl") + raw, err := os.ReadFile(telemSrc) + require.NoError(t, err) + require.NoError(t, os.WriteFile(telem, raw, 0o644)) + reg := filepath.Join("testdata", "sample-registry.yaml") + + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:1") // closed port + results, stats, err := MeasureAndExport(context.Background(), telem, reg, dir, "test-1.2.3") + require.NoError(t, err) + require.NotEmpty(t, results) + _, statErr := os.Stat(filepath.Join(dir, MeasurementsFile)) + require.NoError(t, statErr) + require.NotEmpty(t, stats.RemoteExportWarning, "expected OTLP failure warning with local JSONL kept") + assert.Contains(t, stats.RemoteExportWarning, "otlp export failed for all", + "transport failure must not claim N/M row-construction success") +} + +func TestExportOTLPScores_UnsampledTRACEPARENTNoop(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + // W3C TRACEPARENT with sampled flag cleared (-00). + t.Setenv("TRACEPARENT", "00-84d470ba2451ffeccfe09022d9b2aebd-77f8c0902eaeedcb-00") + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", Label: LabelPass, TraceID: "84d470ba2451ffeccfe09022d9b2aebd", + SpanID: "77f8c0902eaeedcb", Value: 1, Version: "em-001@1", + }}, "test-1.2.3") + require.NoError(t, err) + assert.Empty(t, sink.allSpans(), "unsampled inbound TRACEPARENT must suppress OTLP score export") +} + +func TestExportOTLPScores_SampledTRACEPARENTExports(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + // Sampled inbound parent (-01): common dispatched-pipeline path. + t.Setenv("TRACEPARENT", "00-84d470ba2451ffeccfe09022d9b2aebd-77f8c0902eaeedcb-01") + orig := newScoreOTLPExporter + t.Cleanup(func() { newScoreOTLPExporter = orig }) + newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporter(ctx) + } + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", Label: LabelPass, TraceID: "84d470ba2451ffeccfe09022d9b2aebd", + SpanID: "77f8c0902eaeedcb", Value: 1, Version: "em-001@1", + }}, "test-1.2.3") + require.NoError(t, err) + assert.NotEmpty(t, sink.allSpans(), "sampled TRACEPARENT must not suppress score export") +} + +func TestExportOTLPScores_UnsampledTRACEPARENTOtherTraceIDExports(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + // Unsampled parent for TraceID A must not suppress scores for TraceID B. + t.Setenv("TRACEPARENT", "00-84d470ba2451ffeccfe09022d9b2aebd-77f8c0902eaeedcb-00") + orig := newScoreOTLPExporter + t.Cleanup(func() { newScoreOTLPExporter = orig }) + newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporter(ctx) + } + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", Label: LabelPass, TraceID: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + SpanID: "bbbbbbbbbbbbbbbb", Value: 1, Version: "em-001@1", + }}, "test-1.2.3") + require.NoError(t, err) + assert.NotEmpty(t, sink.allSpans(), "unrelated TraceID must still export under TraceID-scoped gate") +} + +func TestCapturingExporter_ClearsErrorOnLaterSuccess(t *testing.T) { + seq := &seqExporter{errs: []error{assert.AnError, nil}} + cap := &capturingExporter{base: seq} + require.Error(t, cap.ExportSpans(context.Background(), nil)) + cap.mu.Lock() + require.Error(t, cap.err) + cap.mu.Unlock() + require.NoError(t, cap.ExportSpans(context.Background(), nil)) + cap.mu.Lock() + assert.NoError(t, cap.err, "successful ExportSpans must clear prior latch") + cap.mu.Unlock() +} + +type seqExporter struct { + errs []error + i int +} + +func (s *seqExporter) ExportSpans(context.Context, []sdktrace.ReadOnlySpan) error { + if s.i >= len(s.errs) { + return nil + } + err := s.errs[s.i] + s.i++ + return err +} + +func (s *seqExporter) Shutdown(context.Context) error { return nil } + +func TestExportOTLPScores_TruncatesLongExplanation(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + orig := newScoreOTLPExporter + t.Cleanup(func() { newScoreOTLPExporter = orig }) + newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporter(ctx) + } + huge := strings.Repeat("x", telemetry.MaxSpanAttrValueLen+500) + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", Label: LabelPass, Explanation: huge, + TraceID: "84d470ba2451ffeccfe09022d9b2aebd", SpanID: "77f8c0902eaeedcb", + Version: "em-001@1", Value: 1, + }}, "test-1.2.3") + require.NoError(t, err) + got := explanationFromSink(t, sink) + require.NotEmpty(t, got) + assert.LessOrEqual(t, len(got), telemetry.MaxSpanAttrValueLen) + assert.Less(t, len(got), len(huge)) +} + +func TestExportOTLPScores_HonorsOTELAttrValueLimit(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "64") + orig := newScoreOTLPExporter + t.Cleanup(func() { newScoreOTLPExporter = orig }) + newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporter(ctx) + } + huge := strings.Repeat("y", 200) + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", Label: LabelPass, Explanation: huge, + TraceID: "84d470ba2451ffeccfe09022d9b2aebd", SpanID: "77f8c0902eaeedcb", + Version: "em-001@1", Value: 1, + }}, "test-1.2.3") + require.NoError(t, err) + got := explanationFromSink(t, sink) + require.NotEmpty(t, got) + assert.LessOrEqual(t, len([]rune(got)), 64) + assert.Equal(t, 64, len([]rune(got))) +} + +func TestExportOTLPScores_TruncatesUnderContentCapture(t *testing.T) { + // Level 3 content capture lifts SpanLimits to unlimited; event + // explanations must still hit FreeTextAttrValueLenLimit (8192 default). + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "true") + orig := newScoreOTLPExporter + t.Cleanup(func() { newScoreOTLPExporter = orig }) + newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporter(ctx) + } + huge := strings.Repeat("z", telemetry.MaxSpanAttrValueLen+500) + require.Equal(t, -1, telemetry.SpanLimits().AttributeValueLengthLimit, + "precondition: content capture leaves SpanLimits unlimited") + err := ExportOTLPScores(context.Background(), []EvaluationResult{{ + Name: "trace_fitness", Label: LabelPass, Explanation: huge, + TraceID: "84d470ba2451ffeccfe09022d9b2aebd", SpanID: "77f8c0902eaeedcb", + Version: "em-001@1", Value: 1, + }}, "test-1.2.3") + require.NoError(t, err) + got := explanationFromSink(t, sink) + require.NotEmpty(t, got) + assert.Equal(t, telemetry.MaxSpanAttrValueLen, len([]rune(got)), + "explanation must stay capped when Level 3 lifts provider SpanLimits") +} + +func TestExportOTLPScores_PartialIDFailureReportsCounts(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + orig := newScoreOTLPExporter + t.Cleanup(func() { newScoreOTLPExporter = orig }) + newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporter(ctx) + } + err := ExportOTLPScores(context.Background(), []EvaluationResult{ + { + Name: "trace_fitness", Label: LabelPass, + TraceID: "84d470ba2451ffeccfe09022d9b2aebd", SpanID: "77f8c0902eaeedcb", + Version: "em-001@1", Value: 1, + }, + { + Name: "trace_fitness", Label: LabelPass, + TraceID: "not-a-valid-trace-id", SpanID: "77f8c0902eaeedcb", + Version: "em-001@1", Value: 1, + }, + }, "test-1.2.3") + require.Error(t, err) + assert.Contains(t, err.Error(), "1/2 scores exported") + assert.Contains(t, err.Error(), "1 failed") + assert.NotEmpty(t, sink.allSpans(), "good row must still export") +} + +func explanationFromSink(t *testing.T, sink *scoreOTLPSink) string { + t.Helper() + reqs := sink.allSpans() + require.NotEmpty(t, reqs) + for _, req := range reqs { + for _, rs := range req.GetResourceSpans() { + for _, ss := range rs.GetScopeSpans() { + for _, sp := range ss.GetSpans() { + for _, ev := range sp.GetEvents() { + if ev.GetName() != EventGenAIEvaluationResult { + continue + } + for _, kv := range ev.GetAttributes() { + if kv.GetKey() == AttrGenAIEvaluationExplanation { + return kv.GetValue().GetStringValue() + } + } + } + } + } + } + } + return "" +} diff --git a/internal/evalmeasure/parse.go b/internal/evalmeasure/parse.go index 4ec713f079..9915c3103c 100644 --- a/internal/evalmeasure/parse.go +++ b/internal/evalmeasure/parse.go @@ -50,6 +50,9 @@ type ParseStats struct { // but some traces were still recovered. Callers should warn; MeasureAndExport // still scores those traces and returns a nil error. Incomplete string + // RemoteExportWarning is set when portable OTLP score export failed + // after local JSONL persistence. Measurements stay fail-open. + RemoteExportWarning string } // ParseTelemetryFile reads OTLP JSON TracesData lines from run-telemetry.jsonl diff --git a/internal/evalmeasure/run.go b/internal/evalmeasure/run.go index 490c8213a8..7079b60e41 100644 --- a/internal/evalmeasure/run.go +++ b/internal/evalmeasure/run.go @@ -20,13 +20,16 @@ func WithPersistHook(ctx context.Context, fn func()) context.Context { // MeasureFile parses telemetry, scores with the manifest, and writes local // eval-measurements.jsonl. Idempotent per ledger. func MeasureFile(telemetryPath, registryPath, outDir string) ([]EvaluationResult, error) { - r, _, err := MeasureAndExport(context.Background(), telemetryPath, registryPath, outDir) + r, _, err := MeasureAndExport(context.Background(), telemetryPath, registryPath, outDir, "") return r, err } -// MeasureAndExport is MeasureFile with an explicit context (reserved for -// future portable OTLP score export on the same OTEL_* path as ADR 0050). -func MeasureAndExport(ctx context.Context, telemetryPath, registryPath, outDir string) ([]EvaluationResult, ParseStats, error) { +// MeasureAndExport is MeasureFile with an explicit context used for portable +// OTLP score export (same OTEL_EXPORTER_OTLP_* path as ADR 0050). Local +// JSONL/ledger always win; OTLP failures are recorded on ParseStats and +// never fail the measure. serviceVersion should match telemetry.Setup +// (CLI Version()) so remote score resources share agent-trace identity. +func MeasureAndExport(ctx context.Context, telemetryPath, registryPath, outDir, serviceVersion string) ([]EvaluationResult, ParseStats, error) { var stats ParseStats if err := ctx.Err(); err != nil { return nil, stats, err @@ -53,23 +56,38 @@ func MeasureAndExport(ctx context.Context, telemetryPath, registryPath, outDir s var all []EvaluationResult hook, _ := ctx.Value(persistHookKey{}).(func()) + // Fail-open OTLP for any rows already persisted+ledgered, including when + // a later row hits a mid-loop persist error (do not silently drop remote + // export for the successful prefix). + exportScored := func() { + if len(all) == 0 { + return + } + if err := ExportOTLPScores(ctx, all, serviceVersion); err != nil { + stats.RemoteExportWarning = err.Error() + } + } + for _, tr := range traces { results := ScoreTrace(tr, reg) for _, r := range results { done, err := AlreadyScored(ledgerPath, r.TraceID, r.Name, r.Version) if err != nil { + exportScored() return all, stats, fmt.Errorf("check ledger: %w", err) } if done { continue } if err := AppendMeasurements(measPath, []EvaluationResult{r}); err != nil { + exportScored() return all, stats, fmt.Errorf("append measurements: %w", err) } // Only count rows that landed in eval-measurements.jsonl so // CLI stdout matches disk on a later ledger/write error. all = append(all, r) if err := RecordScored(ledgerPath, r.TraceID, r.Name, r.Version); err != nil { + exportScored() return all, stats, fmt.Errorf("record scored: %w", err) } if hook != nil { @@ -77,6 +95,7 @@ func MeasureAndExport(ctx context.Context, telemetryPath, registryPath, outDir s } } } + exportScored() // Partial parse with traces already scored is success: scores are data. // stats.Incomplete (if set) lets the CLI warn without failing the job. return all, stats, nil diff --git a/internal/evalmeasure/run_test.go b/internal/evalmeasure/run_test.go index 7e3566aa86..1d4b1acd86 100644 --- a/internal/evalmeasure/run_test.go +++ b/internal/evalmeasure/run_test.go @@ -9,6 +9,9 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + + "github.com/fullsend-ai/fullsend/internal/telemetry" ) func TestLoadRegistryAndScoreTrace(t *testing.T) { @@ -27,6 +30,7 @@ func TestLoadRegistryAndScoreTrace(t *testing.T) { } func TestMeasureFile_Idempotent(t *testing.T) { + clearOTLPEnv(t) out := t.TempDir() telemetry := filepath.Join("testdata", "complete.jsonl") registry := filepath.Join("testdata", "sample-registry.yaml") @@ -45,6 +49,7 @@ func TestMeasureFile_Idempotent(t *testing.T) { } func TestMeasureFile_AppendBeforeLedger(t *testing.T) { + clearOTLPEnv(t) out := t.TempDir() telemetry := filepath.Join("testdata", "complete.jsonl") registry := filepath.Join("testdata", "sample-registry.yaml") @@ -66,6 +71,7 @@ func TestMeasureFile_AppendBeforeLedger(t *testing.T) { } func TestMeasureFile_BadRegistry(t *testing.T) { + clearOTLPEnv(t) _, err := MeasureFile( filepath.Join("testdata", "complete.jsonl"), filepath.Join(t.TempDir(), "missing.yaml"), @@ -75,6 +81,7 @@ func TestMeasureFile_BadRegistry(t *testing.T) { } func TestMeasureFile_BadTelemetry(t *testing.T) { + clearOTLPEnv(t) _, err := MeasureFile( filepath.Join(t.TempDir(), "missing.jsonl"), filepath.Join("testdata", "sample-registry.yaml"), @@ -84,6 +91,7 @@ func TestMeasureFile_BadTelemetry(t *testing.T) { } func TestMeasureAndExport_CancelledContext(t *testing.T) { + clearOTLPEnv(t) ctx, cancel := context.WithCancel(context.Background()) cancel() _, _, err := MeasureAndExport( @@ -91,6 +99,7 @@ func TestMeasureAndExport_CancelledContext(t *testing.T) { filepath.Join("testdata", "complete.jsonl"), filepath.Join("testdata", "sample-registry.yaml"), t.TempDir(), + "", ) require.Error(t, err) } @@ -114,6 +123,15 @@ func writeTwoTraceTelemetry(t *testing.T, completePath string) string { } func TestMeasureAndExport_KeepsFirstWhenSecondPersistFails(t *testing.T) { + sink := newScoreOTLPSink(t) + clearOTLPEnv(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL+"/v1/traces") + orig := newScoreOTLPExporter + t.Cleanup(func() { newScoreOTLPExporter = orig }) + newScoreOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return telemetry.NewOTLPExporter(ctx) + } + out := t.TempDir() telem := writeTwoTraceTelemetry(t, filepath.Join("testdata", "complete.jsonl")) ctx := WithPersistHook(context.Background(), func() { @@ -121,14 +139,17 @@ func TestMeasureAndExport_KeepsFirstWhenSecondPersistFails(t *testing.T) { require.NoError(t, os.Remove(meas)) require.NoError(t, os.Mkdir(meas, 0o755)) }) - results, _, err := MeasureAndExport(ctx, telem, filepath.Join("testdata", "sample-registry.yaml"), out) + results, stats, err := MeasureAndExport(ctx, telem, filepath.Join("testdata", "sample-registry.yaml"), out, "test-1.2.3") require.Error(t, err) require.Len(t, results, 1) assert.Equal(t, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", results[0].TraceID) assert.Contains(t, err.Error(), "append measurements") + assert.Empty(t, stats.RemoteExportWarning, "persisted prefix must still attempt OTLP export") + require.NotEmpty(t, sink.allSpans(), "first persisted row must be OTLP-exported before mid-loop return") } func TestMeasureAndExport_ScoresPartialFileDespiteParseError(t *testing.T) { + clearOTLPEnv(t) // Oversized line after a good line: ParseTelemetryFile keeps the good // trace and returns sc.Err(); MeasureAndExport still scores it and // treats the partial parse as success (scores are data). @@ -151,7 +172,7 @@ func TestMeasureAndExport_ScoresPartialFileDespiteParseError(t *testing.T) { require.NoError(t, f.Close()) out := t.TempDir() - results, stats, err := MeasureAndExport(context.Background(), telem, filepath.Join("testdata", "sample-registry.yaml"), out) + results, stats, err := MeasureAndExport(context.Background(), telem, filepath.Join("testdata", "sample-registry.yaml"), out, "") require.NoError(t, err) require.Len(t, results, 1) assert.Equal(t, LabelPass, results[0].Label) @@ -160,6 +181,7 @@ func TestMeasureAndExport_ScoresPartialFileDespiteParseError(t *testing.T) { } func TestMeasureFile_PrescriptSkippedRecordsSkip(t *testing.T) { + clearOTLPEnv(t) out := t.TempDir() results, err := MeasureFile( filepath.Join("testdata", "prescript-skipped.jsonl"), @@ -178,6 +200,7 @@ func TestMeasureFile_PrescriptSkippedRecordsSkip(t *testing.T) { } func TestMeasureFile_EmptyIdentityPersistsFailRow(t *testing.T) { + clearOTLPEnv(t) dir := t.TempDir() telem := filepath.Join(dir, "run-telemetry.jsonl") // Minimal OTLP line: run span with no agent identity. diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index d8d5a300d9..4185e3725c 100644 --- a/internal/telemetry/telemetry.go +++ b/internal/telemetry/telemetry.go @@ -40,14 +40,59 @@ const scopeName = "github.com/fullsend-ai/fullsend/internal/telemetry" // newOTLPExporter is a seam over exporter construction for tests. // The SDK reads OTEL_EXPORTER_OTLP_*ENDPOINT from the environment. var newOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return NewOTLPExporter(ctx) +} + +// OTLPEnabled reports whether an OTLP traces endpoint is configured via +// OTEL_EXPORTER_OTLP_ENDPOINT or OTEL_EXPORTER_OTLP_TRACES_ENDPOINT. +func OTLPEnabled() bool { + return strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")) != "" || + strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")) != "" +} + +// NewOTLPExporter builds the HTTP OTLP span exporter from OTEL_* env +// (same path Setup uses for agent traces). Callers must validate endpoints +// first with ValidateOTLPEndpoints when they want fail-closed setup. +func NewOTLPExporter(ctx context.Context) (sdktrace.SpanExporter, error) { + retryOption := otlptracehttp.WithRetry(otlptracehttp.RetryConfig{ + Enabled: true, + InitialInterval: 250 * time.Millisecond, + MaxInterval: 2 * time.Second, + }) + return otlptracehttp.New(ctx, retryOption) +} + +// NewOTLPExporterBounded is NewOTLPExporter with MaxElapsedTime set so +// post-hoc exporters (eval scores) cannot retry forever on a flaky collector. +func NewOTLPExporterBounded(ctx context.Context, maxElapsed time.Duration) (sdktrace.SpanExporter, error) { retryOption := otlptracehttp.WithRetry(otlptracehttp.RetryConfig{ Enabled: true, InitialInterval: 250 * time.Millisecond, MaxInterval: 2 * time.Second, + MaxElapsedTime: maxElapsed, }) return otlptracehttp.New(ctx, retryOption) } +// BuildResource returns the fullsend OTLP resource (service.name, version, +// plus OTEL_RESOURCE_ATTRIBUTES). Shared by agent Setup and score export so +// backends that group by resource keep both on the same service identity. +func BuildResource(serviceVersion string) *resource.Resource { + if serviceVersion == "" { + serviceVersion = "unknown" + } + return buildResource(serviceVersion) +} + +// ValidateOTLPEndpoints checks the OTEL endpoint env vars that the SDK +// will use for traces export. +func ValidateOTLPEndpoints() error { + return validateEndpoints( + strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")), + strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")), + ) +} + func validateEndpoints(endpoint, tracesEndpoint string) error { // The SDK uses TRACES_ENDPOINT when set, falling back to ENDPOINT. // Validate only the value that will actually be used. @@ -99,6 +144,27 @@ func validateEndpoints(endpoint, tracesEndpoint string) error { // there, including -1 (unlimited), is honored as-is. const MaxSpanAttrValueLen = 8192 +// SpanLimits returns the SDK span limits used by Setup (default 8KiB +// attribute value length unless OTEL_*_ATTRIBUTE_VALUE_LENGTH_LIMIT is set, +// or unlimited when Level 3 content capture lifts the provider cap). +func SpanLimits() sdktrace.SpanLimits { + return spanLimits() +} + +// FreeTextAttrValueLenLimit is the call-site bound for free-text values the +// SDK does not truncate (event attributes) or that must stay intact when +// Level 3 content capture lifts SpanLimits. Honors an explicit operator +// OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT / OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT +// (including -1 = unlimited); otherwise defaults to MaxSpanAttrValueLen. +// Independent of ContentCaptureEnabled — that gate only lifts the provider +// cap for content JSON span attrs. +func FreeTextAttrValueLenLimit() int { + if n, ok := operatorAttrValueLimit(); ok { + return n + } + return MaxSpanAttrValueLen +} + // spanLimits returns the SDK span limits. NewSpanLimits collapses "env // unset" and an explicit "-1" (the OTel sentinel for unlimited) to the // same struct value, so the env vars are consulted directly: the diff --git a/internal/telemetry/telemetry_test.go b/internal/telemetry/telemetry_test.go index e69b338121..1697c6c0ce 100644 --- a/internal/telemetry/telemetry_test.go +++ b/internal/telemetry/telemetry_test.go @@ -193,6 +193,24 @@ func TestSpanLimits(t *testing.T) { "a valid specific var wins regardless of the generic one") } +func TestFreeTextAttrValueLenLimit(t *testing.T) { + pinOTELEnv(t) + t.Setenv(ContentCaptureEnvVar, "") + assert.Equal(t, MaxSpanAttrValueLen, FreeTextAttrValueLenLimit()) + + t.Setenv(ContentCaptureEnvVar, "true") + assert.Equal(t, MaxSpanAttrValueLen, FreeTextAttrValueLenLimit(), + "content capture must not lift free-text call-site bound") + assert.Equal(t, -1, spanLimits().AttributeValueLengthLimit, + "precondition: SpanLimits is unlimited under content capture") + + t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "128") + assert.Equal(t, 128, FreeTextAttrValueLenLimit(), "operator finite limit wins") + + t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "-1") + assert.Equal(t, -1, FreeTextAttrValueLenLimit(), "operator -1 stays unlimited") +} + // TestSetup_ContentCaptureOperatorLimitWarning pins the collision warning: // when the Level 3 gate is on but an operator's finite attribute value // length limit is configured, the SDK will cut gen_ai.output.messages @@ -978,3 +996,17 @@ func TestParentSampledProcessor_AllowsSampledTrace(t *testing.T) { assert.ElementsMatch(t, []string{"root", "child"}, spy.ended) } + +func TestOTLPEnabledAndValidate(t *testing.T) { + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") + assert.False(t, OTLPEnabled()) + require.NoError(t, ValidateOTLPEndpoints()) + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:4318") + assert.True(t, OTLPEnabled()) + require.NoError(t, ValidateOTLPEndpoints()) + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "not-a-url") + require.Error(t, ValidateOTLPEndpoints()) +} diff --git a/pkg/behaviourtest/drivers/install/ensure.go b/pkg/behaviourtest/drivers/install/ensure.go index 3df23277b9..e4d8406115 100644 --- a/pkg/behaviourtest/drivers/install/ensure.go +++ b/pkg/behaviourtest/drivers/install/ensure.go @@ -236,6 +236,11 @@ func (e *repoEnsurer) awaitDeletion(ctx context.Context, org, repoName, target s e.logf("[ensure] waiting for %s deletion to propagate", target) delay := resetRetryDelay for attempt := 1; attempt <= resetMaxAttempts; attempt++ { + // Check before poll/backoff: with delay=0 (tests), select on + // time.After(0) vs ctx.Done() is non-deterministic. + if err := ctx.Err(); err != nil { + return fmt.Errorf("context cancelled while waiting for %s deletion: %w", target, err) + } _, err := e.client.GetRepo(ctx, org, repoName) if err != nil { if forge.IsNotFound(err) { @@ -290,6 +295,11 @@ func (e *repoEnsurer) awaitCreation(ctx context.Context, org, repoName, target s e.logf("[ensure] waiting for %s creation to propagate", target) delay := resetRetryDelay for attempt := 1; attempt <= resetMaxAttempts; attempt++ { + // Check before poll/backoff: with delay=0 (tests), select on + // time.After(0) vs ctx.Done() is non-deterministic. + if err := ctx.Err(); err != nil { + return fmt.Errorf("context cancelled while waiting for %s creation: %w", target, err) + } _, err := e.client.GetRepo(ctx, org, repoName) if err == nil { e.logf("[ensure] %s creation confirmed after %d attempt(s)", target, attempt)