diff --git a/CHANGELOG.md b/CHANGELOG.md index c902cac..c068e55 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,16 @@ Notable user-visible changes are recorded here, newest first. -## [0.13.0] - Unreleased +## [Unreleased] + +- Read only Langfuse's bounded `level` enum, show ERROR observation token + subsets and avoid severity-only usage-review flags for DEFAULT/DEBUG/WARNING. + Provider call status and billing finality remain unknown; missing/invalid + levels still require review. Token/cost totals and LiteLLM behavior are unchanged. +- Extend JSON/JSONL/CSV and gzip privacy checks without reading `status_message` + or observation payloads. This does not change the published v0.13.0 artifacts. + +## [0.13.0] - 2026-10-24 See the [release notes](docs/releases/v0.13.0.md). diff --git a/docs/LANGFUSE.md b/docs/LANGFUSE.md index 00394ba..c90ca35 100644 --- a/docs/LANGFUSE.md +++ b/docs/LANGFUSE.md @@ -57,9 +57,11 @@ Who drove the increase? 1 leading tracked trace_name accounts for 100% of the net recorded increase. ``` -This version doesn't read a success or status field from Langfuse observations, -so the report marks status as unknown and lists those records for usage review. -Recorded token and cost totals are still shown. +The v0.13.0 binary does not read a success or status field from Langfuse +observations: it reports unknown status and lists those records for usage review. +Recorded token and cost totals are still shown. The fixture's `level` field is +ignored by v0.13.0; non-error severity is not proof of provider success or complete +billing. ## Grouping and accounting diff --git a/examples/langfuse-spend/export/synthetic-langfuse/manifests/2026-10-07T01-00-00.json b/examples/langfuse-spend/export/synthetic-langfuse/manifests/2026-10-07T01-00-00.json index 3cf49e6..7395dfb 100644 --- a/examples/langfuse-spend/export/synthetic-langfuse/manifests/2026-10-07T01-00-00.json +++ b/examples/langfuse-spend/export/synthetic-langfuse/manifests/2026-10-07T01-00-00.json @@ -15,7 +15,7 @@ "format": "jsonl-raw", "compressed": false, "contentType": "application/x-ndjson; charset=utf-8", - "sizeBytes": 1322, + "sizeBytes": 1394, "rowCount": 4 } ] diff --git a/examples/langfuse-spend/export/synthetic-langfuse/manifests/2026-10-08T01-00-00.json b/examples/langfuse-spend/export/synthetic-langfuse/manifests/2026-10-08T01-00-00.json index 9a42058..5c14cea 100644 --- a/examples/langfuse-spend/export/synthetic-langfuse/manifests/2026-10-08T01-00-00.json +++ b/examples/langfuse-spend/export/synthetic-langfuse/manifests/2026-10-08T01-00-00.json @@ -15,7 +15,7 @@ "format": "jsonl-raw", "compressed": false, "contentType": "application/x-ndjson; charset=utf-8", - "sizeBytes": 1984, + "sizeBytes": 2092, "rowCount": 6 } ] diff --git a/examples/langfuse-spend/export/synthetic-langfuse/observations_v2/2026-10-07T00-00-00.jsonl b/examples/langfuse-spend/export/synthetic-langfuse/observations_v2/2026-10-07T00-00-00.jsonl index 179d589..6f7a0ed 100644 --- a/examples/langfuse-spend/export/synthetic-langfuse/observations_v2/2026-10-07T00-00-00.jsonl +++ b/examples/langfuse-spend/export/synthetic-langfuse/observations_v2/2026-10-07T00-00-00.jsonl @@ -1,4 +1,4 @@ -{"id":"2026-10-07-1","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-07 00:05:00.000000","end_time":"2026-10-07 00:05:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} -{"id":"2026-10-07-2","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-07 00:10:00.000000","end_time":"2026-10-07 00:10:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} -{"id":"2026-10-07-3","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-07 00:15:00.000000","end_time":"2026-10-07 00:15:01.000000","provided_model_name":"synthetic-model","trace_name":"background","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} -{"id":"2026-10-07-4","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-07 00:20:00.000000","end_time":"2026-10-07 00:20:01.000000","provided_model_name":"synthetic-model","trace_name":"background","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} +{"level":"DEFAULT","id":"2026-10-07-1","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-07 00:05:00.000000","end_time":"2026-10-07 00:05:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} +{"level":"DEFAULT","id":"2026-10-07-2","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-07 00:10:00.000000","end_time":"2026-10-07 00:10:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} +{"level":"DEFAULT","id":"2026-10-07-3","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-07 00:15:00.000000","end_time":"2026-10-07 00:15:01.000000","provided_model_name":"synthetic-model","trace_name":"background","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} +{"level":"DEFAULT","id":"2026-10-07-4","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-07 00:20:00.000000","end_time":"2026-10-07 00:20:01.000000","provided_model_name":"synthetic-model","trace_name":"background","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} diff --git a/examples/langfuse-spend/export/synthetic-langfuse/observations_v2/2026-10-08T00-00-00.jsonl b/examples/langfuse-spend/export/synthetic-langfuse/observations_v2/2026-10-08T00-00-00.jsonl index efbc8e5..6965c57 100644 --- a/examples/langfuse-spend/export/synthetic-langfuse/observations_v2/2026-10-08T00-00-00.jsonl +++ b/examples/langfuse-spend/export/synthetic-langfuse/observations_v2/2026-10-08T00-00-00.jsonl @@ -1,6 +1,6 @@ -{"id":"2026-10-08-1","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:05:00.000000","end_time":"2026-10-08 00:05:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":2250,"output":750,"total":3000},"total_cost":0.003} -{"id":"2026-10-08-2","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:10:00.000000","end_time":"2026-10-08 00:10:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":2250,"output":750,"total":3000},"total_cost":0.003} -{"id":"2026-10-08-3","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:15:00.000000","end_time":"2026-10-08 00:15:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":2250,"output":750,"total":3000},"total_cost":0.003} -{"id":"2026-10-08-4","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:20:00.000000","end_time":"2026-10-08 00:20:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":2250,"output":750,"total":3000},"total_cost":0.003} -{"id":"2026-10-08-5","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:25:00.000000","end_time":"2026-10-08 00:25:01.000000","provided_model_name":"synthetic-model","trace_name":"background","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} -{"id":"2026-10-08-6","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:30:00.000000","end_time":"2026-10-08 00:30:01.000000","provided_model_name":"synthetic-model","trace_name":"background","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} +{"level":"DEFAULT","id":"2026-10-08-1","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:05:00.000000","end_time":"2026-10-08 00:05:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":2250,"output":750,"total":3000},"total_cost":0.003} +{"level":"DEFAULT","id":"2026-10-08-2","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:10:00.000000","end_time":"2026-10-08 00:10:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":2250,"output":750,"total":3000},"total_cost":0.003} +{"level":"DEFAULT","id":"2026-10-08-3","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:15:00.000000","end_time":"2026-10-08 00:15:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":2250,"output":750,"total":3000},"total_cost":0.003} +{"level":"DEFAULT","id":"2026-10-08-4","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:20:00.000000","end_time":"2026-10-08 00:20:01.000000","provided_model_name":"synthetic-model","trace_name":"interactive","release":"synthetic-example","usage_details":{"input":2250,"output":750,"total":3000},"total_cost":0.003} +{"level":"DEFAULT","id":"2026-10-08-5","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:25:00.000000","end_time":"2026-10-08 00:25:01.000000","provided_model_name":"synthetic-model","trace_name":"background","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} +{"level":"DEFAULT","id":"2026-10-08-6","project_id":"synthetic-langfuse","type":"GENERATION","start_time":"2026-10-08 00:30:00.000000","end_time":"2026-10-08 00:30:01.000000","provided_model_name":"synthetic-model","trace_name":"background","release":"synthetic-example","usage_details":{"input":1500,"output":500,"total":2000},"total_cost":0.002} diff --git a/internal/cli/langfuse.go b/internal/cli/langfuse.go index d4576eb..161f642 100644 --- a/internal/cli/langfuse.go +++ b/internal/cli/langfuse.go @@ -46,6 +46,7 @@ func runLangfuse(dir, group, tag, format string, opts spend.Options, out, errout text = strings.ReplaceAll(text, "Comparison incomplete: some selected call types are not analyzed.", "Comparison incomplete: the supplied local export does not fully cover the selected periods.") text = strings.ReplaceAll(text, "gateway-recorded", "Langfuse-recorded") text = strings.ReplaceAll(text, "your local LiteLLM logs", "your local Langfuse export") + text = strings.ReplaceAll(text, "Unknown status", "Call status unknown (Langfuse doesn't record it)") if c.Incomplete { text = strings.ReplaceAll(text, "Whole-period per-request split unavailable while selected call types remain unanalyzed.", "Whole-period per-request split unavailable while local export coverage is incomplete.") text = strings.ReplaceAll(text, "Next: review unsupported call types in your local Langfuse export; analyze those rows before drawing a whole-export token-change conclusion.", "Next: restore missing or partial files and manifests before drawing a whole-period conclusion; observed totals above remain partial.") diff --git a/internal/cli/langfuse_docs_test.go b/internal/cli/langfuse_docs_test.go index 1b36407..3ae9fd2 100644 --- a/internal/cli/langfuse_docs_test.go +++ b/internal/cli/langfuse_docs_test.go @@ -40,7 +40,8 @@ func checkLangfuseExample(t *testing.T, format string, output []byte) { r.Before.Requests != 4 || r.After.Requests != 6 || r.Before.Tokens != 8000 || r.After.Tokens != 16000 || r.Before.RecordedSpend != "0.008" || r.After.RecordedSpend != "0.016" || r.Before.Quality.UnknownStatus != 4 || r.After.Quality.UnknownStatus != 6 || - r.Before.Quality.Unclear != 4 || r.After.Quality.Unclear != 6 || + r.Before.Quality.Unclear != 0 || r.After.Quality.Unclear != 0 || + r.Before.Quality.ObservationLevels["DEFAULT"] != 4 || r.After.Quality.ObservationLevels["DEFAULT"] != 6 || r.Increase == nil || r.Increase.Count != 1 || r.Increase.Share.Lower != 1 { t.Fatal("public example totals, contributor or manifest coverage differ") } @@ -49,11 +50,15 @@ func checkLangfuseExample(t *testing.T, format string, output []byte) { "trace_name-1: 4,000 -> 12,000 tokens; change +8,000.", "trace_name-2: 4,000 -> 4,000 tokens; change +0.", "1 leading tracked trace_name accounts for 100% of the net recorded increase.", - "Unknown status: 4 -> 6."} { + "Call status unknown (Langfuse doesn't record it): 4 -> 6.", + "Observation levels: DEFAULT 4 -> 6."} { if !strings.Contains(string(output), want) { t.Fatalf("example missing %q", want) } } + if strings.Contains(string(output), "Records needing usage review") || strings.Contains(string(output), "Tokens recorded on ERROR observations") { + t.Fatal("clean DEFAULT example contains review or ERROR-token noise") + } } } @@ -62,9 +67,12 @@ func TestLangfuseGuideStatusContract(t *testing.T) { if err != nil { t.Fatal(err) } - want := "This version doesn't read a success or status field from Langfuse observations, so the report marks status as unknown and lists those records for usage review. Recorded token and cost totals are still shown." + want := "non-error severity is not proof of provider success or complete billing." if !strings.Contains(strings.Join(strings.Fields(string(doc)), " "), want) { - t.Fatal("Langfuse status limitation must describe this version, not only the example") + t.Fatal("Langfuse severity must not imply provider success or billing completeness") + } + if strings.Contains(string(doc), "## Observation severity") || strings.Contains(string(doc), "(unreleased)") { + t.Fatal("deferred next-release section must not ship on main yet") } } diff --git a/internal/cli/langfuse_level_test.go b/internal/cli/langfuse_level_test.go new file mode 100644 index 0000000..8dec61e --- /dev/null +++ b/internal/cli/langfuse_level_test.go @@ -0,0 +1,163 @@ +// SPDX-License-Identifier: Apache-2.0 +// Code authors: Vijay and Codex + +package cli + +import ( + "bytes" + "context" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/llm-measurement/fleetdiff/internal/langfuse" + "github.com/llm-measurement/fleetdiff/internal/spend" +) + +func TestLangfuseLevelCLIReviewPrivacyAndSubsets(t *testing.T) { + testLangfuseLevel(t, "") +} + +func TestReleaseBinaryLangfuseLevel(t *testing.T) { + binary := os.Getenv("FLEETDIFF_RELEASE_BINARY") + if binary == "" { + t.Skip("set FLEETDIFF_RELEASE_BINARY to test the packaged binary") + } + binary, err := filepath.Abs(binary) + if err != nil { + t.Fatal(err) + } + testLangfuseLevel(t, binary) +} + +func testLangfuseLevel(t *testing.T, binary string) { + t.Helper() + for _, mode := range []string{"known-non-error", "error-and-invalid"} { + t.Run(mode, func(t *testing.T) { + root := t.TempDir() + if err := os.CopyFS(root, os.DirFS("../../examples/langfuse-spend/export")); err != nil { + t.Fatal(err) + } + files, err := filepath.Glob(filepath.Join(root, "synthetic-langfuse/observations_v2/*.jsonl")) + if err != nil || len(files) != 2 { + t.Fatal("synthetic files unavailable") + } + for _, file := range files { + data, err := os.ReadFile(file) + if err != nil { + t.Fatal(err) + } + var body strings.Builder + for i, line := range strings.Split(strings.TrimSpace(string(data)), "\n") { + line = strings.Replace(line, `"level":"DEFAULT",`, "", 1) + level := []string{"DEFAULT", "DEBUG", "WARNING"}[i%3] + if mode == "error-and-invalid" && i == 0 { + level = "ERROR" + } else if mode == "error-and-invalid" && i == 1 { + level = "PRIVATE_LEVEL_SENTINEL" + } + body.WriteString(`{"level":"` + level + `","input":"PRIVATE_INPUT_SENTINEL","output":"PRIVATE_OUTPUT_SENTINEL","metadata":"PRIVATE_METADATA_SENTINEL","tool_calls":"PRIVATE_TOOLS_SENTINEL","status_message":"PRIVATE_STATUS_SENTINEL",` + line[1:] + "\n") + } + if err := os.WriteFile(file, []byte(body.String()), 0600); err != nil { + t.Fatal(err) + } + } + manifests, err := filepath.Glob(filepath.Join(root, "synthetic-langfuse/manifests/*.json")) + if err != nil || len(manifests) != 2 { + t.Fatal("synthetic manifests unavailable") + } + for _, path := range manifests { + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var m map[string]any + if err := json.Unmarshal(data, &m); err != nil { + t.Fatal(err) + } + for _, f := range m["files"].([]any) { + file := f.(map[string]any) + info, err := os.Stat(filepath.Join(root, "synthetic-langfuse/observations_v2", filepath.Base(file["key"].(string)))) + if err != nil { + t.Fatal(err) + } + file["sizeBytes"] = info.Size() + } + data, _ = json.Marshal(m) + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + } + for _, format := range []string{"text", "json"} { + args := langfuseExampleArgs(format) + args[2] = root + var out, diagnostic bytes.Buffer + code := 0 + if binary == "" { + code = Run(args, &out, &diagnostic) + } else { + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + cmd := exec.CommandContext(ctx, binary, args...) + cmd.Stdout, cmd.Stderr = &out, &diagnostic + err := cmd.Run() + cancel() + if err != nil { + t.Fatal("packaged severity command failed", err) + } + } + if code != 0 || diagnostic.Len() != 0 { + t.Fatal("severity command failed", code, diagnostic.String()) + } + if strings.Contains(out.String()+diagnostic.String(), "PRIVATE_") { + t.Fatal("severity or payload sentinel leaked") + } + if format == "text" { + completeness := strings.Index(out.String(), "How complete is the export?") + levels := strings.Index(out.String(), " Observation levels:") + if completeness < 0 || levels < completeness || strings.Count(out.String(), "Observation levels:") != 1 || strings.Contains(out.String(), "Observation level DEFAULT:") || strings.Contains(out.String(), "unknown 0 -> 0") || strings.Contains(out.String(), "ERROR 0 -> 0") { + t.Fatal("level summary is not compact, nonzero-only and in completeness section") + } + if !strings.Contains(out.String(), "provider call status is unknown") || mode == "error-and-invalid" && !strings.Contains(out.String(), "Tokens recorded on ERROR observations:") { + t.Fatal("severity/status or error-subset wording absent") + } + if mode == "known-non-error" && strings.Contains(out.String(), "Records needing usage review") { + t.Fatal("non-error severity caused usage-review noise") + } + if mode == "known-non-error" && strings.Contains(out.String(), "Tokens recorded on ERROR observations:") { + t.Fatal("zero ERROR-token line should be omitted") + } + if !strings.Contains(out.String(), "Call status unknown (Langfuse doesn't record it): 4 -> 6.") { + t.Fatal("short call-status wording absent") + } + continue + } + var r langfuse.Report + if err := json.Unmarshal(out.Bytes(), &r); err != nil { + t.Fatal(err) + } + if r.Before.Tokens != 8000 || r.After.Tokens != 16000 || r.Before.RecordedSpend != "0.008" || r.After.RecordedSpend != "0.016" || r.Before.Quality.UnknownStatus != 4 || r.After.Quality.UnknownStatus != 6 || r.Before.Quality.Failed != 0 || r.After.Quality.Failed != 0 || r.Volume != nil { + t.Fatal("level changed recorded totals or claimed provider status") + } + if mode == "known-non-error" { + if r.Before.Quality.Unclear != 0 || r.After.Quality.Unclear != 0 || r.Before.ErrorObservationTokens != nil { + t.Fatal("non-error severity caused review/error subset") + } + } else if r.Before.Quality.Unclear != 2 || r.After.Quality.Unclear != 2 || r.Before.Quality.ErrorObservations != 1 || r.After.Quality.ObservationLevels["unknown"] != 1 || r.Before.ErrorObservationTokens == nil || r.Before.ErrorObservationTokens.Total != 2000 || r.After.ErrorObservationTokens == nil || r.After.ErrorObservationTokens.Total != 3000 { + t.Fatal("error or invalid severity review/subset differs") + } + } + }) + } +} + +func TestZeroErrorTokenLineOmitted(t *testing.T) { + var out bytes.Buffer + renderSpend(&out, spend.Report{Before: spend.Window{ErrorObservationTokens: &spend.FailureTokens{Finality: "unknown"}}}) + if strings.Contains(out.String(), "Tokens recorded on ERROR observations:") { + t.Fatal("non-nil zero token subset produced an ERROR-token line") + } +} diff --git a/internal/cli/spend.go b/internal/cli/spend.go index 39ac76b..22ce7f3 100644 --- a/internal/cli/spend.go +++ b/internal/cli/spend.go @@ -199,6 +199,25 @@ func renderSpend(out io.Writer, r spend.Report) { } } a, b := r.Before.Quality, r.After.Quality + var levels []string + for _, level := range []string{"DEFAULT", "DEBUG", "WARNING", "ERROR", "unknown"} { + if a.ObservationLevels[level] != 0 || b.ObservationLevels[level] != 0 { + levels = append(levels, fmt.Sprintf("%s %s -> %s", level, spendNumber(a.ObservationLevels[level]), spendNumber(b.ObservationLevels[level]))) + } + } + if len(levels) != 0 { + fmt.Fprintln(out, " Observation levels: "+strings.Join(levels, "; ")+".") + } + errorBefore, errorAfter := spend.FailureTokens{}, spend.FailureTokens{} + if r.Before.ErrorObservationTokens != nil { + errorBefore = *r.Before.ErrorObservationTokens + } + if r.After.ErrorObservationTokens != nil { + errorAfter = *r.After.ErrorObservationTokens + } + if errorBefore.Total != 0 || errorAfter.Total != 0 { + fmt.Fprintf(out, " Tokens recorded on ERROR observations: %s -> %s (input %s -> %s; output %s -> %s); included in totals, provider completion and billing finality unknown.\n", spendNumber(errorBefore.Total), spendNumber(errorAfter.Total), spendNumber(errorBefore.Input), spendNumber(errorAfter.Input), spendNumber(errorBefore.Output), spendNumber(errorAfter.Output)) + } for _, v := range []struct { label string a, b uint64 @@ -255,6 +274,8 @@ func renderSpend(out io.Writer, r spend.Report) { fmt.Fprintln(out, ".") if r.Incomplete { fmt.Fprintln(out, " Whole-period per-request split unavailable while selected call types remain unanalyzed.") + } else if a.ObservationLevels != nil || b.ObservationLevels != nil { + fmt.Fprintln(out, " Whole-period per-request split unavailable while provider call status is unknown; observation severity does not establish success.") } else if r.Volume == nil { fmt.Fprintln(out, " Whole-period per-request split needs clearer usage; recorded totals remain shown.") } @@ -264,6 +285,8 @@ func renderSpend(out io.Writer, r spend.Report) { } if m.Before.Requests == 0 || m.After.Requests == 0 { fmt.Fprintf(out, " %s per-request split needs analyzed requests in both periods.\n", m.Item) + } else if m.Before.Quality.ObservationLevels != nil || m.After.Quality.ObservationLevels != nil { + fmt.Fprintf(out, " %s per-request split unavailable while provider call status is unknown.\n", m.Item) } else { fmt.Fprintf(out, " %s per-request split needs clearer usage on %s before / %s after records.\n", m.Item, spendNumber(m.Before.Quality.Unclear), spendNumber(m.After.Quality.Unclear)) } diff --git a/internal/cli/testdata/langfuse/json.golden b/internal/cli/testdata/langfuse/json.golden index 630a1b3..0acf7aa 100644 --- a/internal/cli/testdata/langfuse/json.golden +++ b/internal/cli/testdata/langfuse/json.golden @@ -156,6 +156,9 @@ "zero_only_origin_unknown": 0, "failed_records": 0, "unknown_status": 5, + "observation_levels": { + "unknown": 5 + }, "unclear_usage_records": 5, "missing_group_identity": 0, "missing_recorded_spend": 0, @@ -219,6 +222,9 @@ "zero_only_origin_unknown": 0, "failed_records": 0, "unknown_status": 5, + "observation_levels": { + "unknown": 5 + }, "unclear_usage_records": 5, "missing_group_identity": 0, "missing_recorded_spend": 0, @@ -285,6 +291,9 @@ "zero_only_origin_unknown": 0, "failed_records": 0, "unknown_status": 5, + "observation_levels": { + "unknown": 5 + }, "unclear_usage_records": 5, "missing_group_identity": 0, "missing_recorded_spend": 0, @@ -345,6 +354,9 @@ "zero_only_origin_unknown": 0, "failed_records": 0, "unknown_status": 5, + "observation_levels": { + "unknown": 5 + }, "unclear_usage_records": 5, "missing_group_identity": 0, "missing_recorded_spend": 0, @@ -410,7 +422,8 @@ "Native input, cache-read and cache-creation are disjoint; inclusive input is their sum. Inclusive output includes disjoint reasoning.", "Recorded cost (Langfuse) is the exact sum of usable total_cost values, not an invoice. Zero cost with usage has unknown origin.", "Manifest coverage validates the supplied local copy, not upstream ingestion or hidden provider retries. Missing/partial files are not zero usage.", - "Raw user, session and custom labels are suppressed with local keyed aliases. Status is unknown; status_message is never decoded." + "Raw user, session and custom labels are suppressed with local keyed aliases. Status is unknown; status_message is never decoded.", + "Level is an observation severity enum, not provider success or billing finality. DEFAULT/DEBUG/WARNING alone do not require usage review; ERROR and missing/invalid levels do. Other usage checks still apply. Per-request splits remain unavailable while call status is unknown." ], "langfuse_coverage": { "manifests": 2, diff --git a/internal/cli/testdata/langfuse/text.golden b/internal/cli/testdata/langfuse/text.golden index a06a973..fb55e48 100644 --- a/internal/cli/testdata/langfuse/text.golden +++ b/internal/cli/testdata/langfuse/text.golden @@ -28,16 +28,17 @@ Whose uncached input grew? Excluded: missing detail 1 -> 1 requests. How complete is the export? + Observation levels: unknown 5 -> 5. Records needing usage review (categories may overlap): 5 -> 5. Missing token fields: 1 -> 1. Total field absent: 1 -> 1. - Unknown status: 5 -> 5. + Call status unknown (Langfuse doesn't record it): 5 -> 5. Zero-filled: cannot determine from this export. Provider origin: not established by Langfuse observations. Recorded cost (Langfuse, analyzed GENERATION observations): 0.0003 -> 0.0003. Zero-cost origin unknown: 4 -> 4. No other usage problems found. Rows read: 28; unsupported rows: 18. - Whole-period per-request split needs clearer usage; recorded totals remain shown. - model-1 per-request split needs clearer usage on 5 before / 5 after records. + Whole-period per-request split unavailable while provider call status is unknown; observation severity does not establish success. + model-1 per-request split unavailable while provider call status is unknown. Next: review the leading contributors and flagged records in your local Langfuse export. diff --git a/internal/langfuse/csv.go b/internal/langfuse/csv.go index 7bd8e98..1af92b5 100644 --- a/internal/langfuse/csv.go +++ b/internal/langfuse/csv.go @@ -13,10 +13,14 @@ import ( // csvField scans RFC4180 quoting even for discarded columns. Only approved // columns have a value buffer; encoding/csv cannot provide that boundary. -func (l *lexer) csvField(keep bool) ([]byte, byte, error) { +func (l *lexer) csvField(keep, level bool) ([]byte, byte, error) { var out []byte add := func(b byte) error { if keep { + if level && len(out) == 7 { + out, keep = nil, false + return nil + } if len(out) >= MaxFieldBytes { return ErrInvalid } @@ -107,8 +111,9 @@ func (l *lexer) csvRecord(names []string) ([][]byte, error) { if names != nil && i >= len(names) { return nil, ErrInvalid } - keep := names == nil || rowKeys[names[i]] - value, end, err := l.csvField(keep) + level := names != nil && names[i] == "level" + keep := names == nil || rowKeys[names[i]] || level + value, end, err := l.csvField(keep, level) if err != nil { return nil, err } @@ -154,6 +159,10 @@ func readCSV(r io.Reader, visit func(projection) error) error { } p := projection{fields: map[string][]byte{}} for i, name := range names { + if name == "level" { + p.level = observationLevel(string(values[i])) + continue + } if !rowKeys[name] { continue } diff --git a/internal/langfuse/directory.go b/internal/langfuse/directory.go index b2af928..ad1e152 100644 --- a/internal/langfuse/directory.go +++ b/internal/langfuse/directory.go @@ -28,6 +28,11 @@ const maxDirectoryBytes = 256 << 20 const maxProjectionBytes = 64 << 20 const maxDirectoryFiles = 4096 +var duplicateFields = [...]string{"level", "type", "start_time", "end_time", "total_cost", "usage_details", "provided_model_name", "user_id", "session_id", "environment", "release", "trace_name", "prompt_name", "tags", "accounting/grouping"} + +// Fixed field hashes name a conflict without retaining or printing its values. +type projectionFingerprint [len(duplicateFields)][sha256.Size]byte + type Coverage struct { Manifests int `json:"manifests"` Projects int `json:"projects"` @@ -218,7 +223,7 @@ func loadDirectory(dir, group, tag string) (snapshot, error) { } out.coverage.Manifests = len(runs) listed := map[string]bool{} - seen := map[[32]byte][32]byte{} + seen := map[[32]byte]projectionFingerprint{} cost := new(big.Rat) projectionBytes := 0 for _, run := range runs { @@ -285,7 +290,11 @@ func loadDirectory(dir, group, tag string) (snapshot, error) { } if prior, ok := seen[id]; ok { if prior != fingerprint { - return out, errors.New("conflicting duplicate Langfuse observation projection") + for i, field := range duplicateFields { + if prior[i] != fingerprint[i] { + return out, errors.New("conflicting duplicate Langfuse observation field: " + field) + } + } } out.coverage.Duplicates++ continue @@ -299,6 +308,7 @@ func loadDirectory(dir, group, tag string) (snapshot, error) { for k, v := range row.Values { projectionBytes += len(k) + len(v) } + projectionBytes += len(row.ObservationLevel) if projectionBytes > maxProjectionBytes { return out, ErrInvalid } @@ -371,16 +381,16 @@ func nativeTime(s string) (time.Time, error) { func framed(parts ...string) string { b, _ := json.Marshal(parts); return string(b) } -func mappedRow(p projection, group, tag string) (spend.Row, [32]byte, [32]byte, error) { - row := spend.Row{Values: map[string]string{}} +func mappedRow(p projection, group, tag string) (spend.Row, [32]byte, projectionFingerprint, error) { + row := spend.Row{Values: map[string]string{}, ObservationLevel: observationLevel(p.level)} v := row.Values project, e := text(p.fields["project_id"]) if e != nil { - return row, [32]byte{}, [32]byte{}, e + return row, [32]byte{}, projectionFingerprint{}, e } id, e := text(p.fields["id"]) if e != nil { - return row, [32]byte{}, [32]byte{}, e + return row, [32]byte{}, projectionFingerprint{}, e } digest := sha256.Sum256([]byte(framed(project, id))) v["request_id"] = hex.EncodeToString(digest[:]) @@ -388,14 +398,14 @@ func mappedRow(p projection, group, tag string) (spend.Row, [32]byte, [32]byte, if s, err := text(p.fields[native]); err == nil { t, err := nativeTime(s) if err != nil { - return row, digest, [32]byte{}, err + return row, digest, projectionFingerprint{}, err } v[column] = t.Format(time.RFC3339Nano) } } kind, err := text(p.fields["type"]) if err != nil { - return row, digest, [32]byte{}, err + return row, digest, projectionFingerprint{}, err } v["call_type"] = "langfuse_other" if kind == "GENERATION" { @@ -419,7 +429,7 @@ func mappedRow(p projection, group, tag string) (spend.Row, [32]byte, [32]byte, if ok { sum, err = accounting.AddBounded(sum, n) if err != nil { - return row, digest, [32]byte{}, ErrAccountingLimit + return row, digest, projectionFingerprint{}, ErrAccountingLimit } } } @@ -455,7 +465,7 @@ func mappedRow(p projection, group, tag string) (spend.Row, [32]byte, [32]byte, var tags []string raw := p.fields["tags"] if len(raw) > 0 && string(raw) != "null" && json.Unmarshal(raw, &tags) != nil { - return row, digest, [32]byte{}, ErrInvalid + return row, digest, projectionFingerprint{}, ErrInvalid } if slices.Contains(tags, tag) { label = tag @@ -467,7 +477,7 @@ func mappedRow(p projection, group, tag string) (spend.Row, [32]byte, [32]byte, } } // Compare normalized approved fields, not encoding-specific raw JSON/CSV. - canonical := map[string]any{"values": v, "kind": kind, "unknown": p.unknown, "bad_usage": p.badUsage} + canonical := map[string]any{"accounting/grouping": v, "type": kind, "level": row.ObservationLevel, "start_time": v["startTime"], "end_time": v["endTime"], "total_cost": v["spend"]} for _, name := range []string{"user_id", "session_id", "provided_model_name", "environment", "release", "trace_name", "prompt_name"} { s, _ := text(p.fields[name]) canonical[name] = s @@ -481,11 +491,11 @@ func mappedRow(p projection, group, tag string) (spend.Row, [32]byte, [32]byte, usage[name] = "invalid" } } - canonical["usage"] = usage + canonical["usage_details"] = map[string]any{"components": usage, "unknown": p.unknown, "bad_usage": p.badUsage} var tags []string if raw := p.fields["tags"]; len(raw) > 0 && string(raw) != "null" { if json.Unmarshal(raw, &tags) != nil { - return row, digest, [32]byte{}, ErrInvalid + return row, digest, projectionFingerprint{}, ErrInvalid } } slices.Sort(tags) @@ -494,8 +504,12 @@ func mappedRow(p projection, group, tag string) (spend.Row, [32]byte, [32]byte, tags = []string{} } canonical["tags"] = tags - data, _ := json.Marshal(canonical) - return row, digest, sha256.Sum256(data), nil + var fingerprint projectionFingerprint + for i, field := range duplicateFields { + data, _ := json.Marshal(canonical[field]) + fingerprint[i] = sha256.Sum256(data) + } + return row, digest, fingerprint, nil } func covered(windows []spend.Period, p spend.Period) bool { @@ -564,6 +578,7 @@ func Investigate(dir, group, tag string, opts spend.Options) (Report, error) { "Recorded cost (Langfuse) is the exact sum of usable total_cost values, not an invoice. Zero cost with usage has unknown origin.", "Manifest coverage validates the supplied local copy, not upstream ingestion or hidden provider retries. Missing/partial files are not zero usage.", "Raw user, session and custom labels are suppressed with local keyed aliases. Status is unknown; status_message is never decoded.", + "Level is an observation severity enum, not provider success or billing finality. DEFAULT/DEBUG/WARNING alone do not require usage review; ERROR and missing/invalid levels do. Other usage checks still apply. Per-request splits remain unavailable while call status is unknown.", } for _, c := range []*compare.Concentration{&r.Rankings, r.Cache.Rankings} { if c != nil { diff --git a/internal/langfuse/directory_test.go b/internal/langfuse/directory_test.go index f471143..92c734e 100644 --- a/internal/langfuse/directory_test.go +++ b/internal/langfuse/directory_test.go @@ -173,7 +173,7 @@ func TestDirectoryCoverageFailures(t *testing.T) { func TestNormalizedDuplicateProjection(t *testing.T) { var ids [][32]byte - var fingerprints [][32]byte + var fingerprints []projectionFingerprint for _, body := range []string{rowJSON("id", "project", "2026-10-10"), strings.ReplaceAll(rowJSON("id", "project", "2026-10-10"), `"total_cost":0.01`, `"total_cost":1e-2`)} { err := readJSON(strings.NewReader(body), false, func(p projection) error { _, id, fingerprint, err := mappedRow(p, "user_id", "") diff --git a/internal/langfuse/json.go b/internal/langfuse/json.go index 6985209..cc8c64d 100644 --- a/internal/langfuse/json.go +++ b/internal/langfuse/json.go @@ -5,6 +5,7 @@ package langfuse import ( "bufio" + "bytes" "encoding/json" "io" "unicode/utf8" @@ -292,11 +293,38 @@ var rowKeys = map[string]bool{ type projection struct { fields map[string][]byte + level string usage map[string][]byte unknown uint64 badUsage bool } +func observationLevel(s string) string { + switch s { + case "DEFAULT", "DEBUG", "WARNING", "ERROR": + return s + default: + return "unknown" + } +} + +// Seven ASCII characters, even when all are JSON-escaped, fit in 44 bytes. +// Decode only that bounded lookahead; scan the full value without retaining it. +func (l *lexer) level() (string, error) { + if l.space() != nil { + return "unknown", ErrInvalid + } + level := "unknown" + if l.peek() == '"' { + raw, _ := l.r.Peek(44) + var s string + if json.NewDecoder(bytes.NewReader(raw)).Decode(&s) == nil { + level = observationLevel(s) + } + } + return level, l.value(nil, 1) +} + func (l *lexer) usage(p *projection) error { p.usage = map[string][]byte{} if l.space() != nil { @@ -379,7 +407,13 @@ func (l *lexer) object() (projection, error) { if l.space() != nil || l.expect(':', nil) != nil { return p, ErrInvalid } - if key == "usage_details" { + if key == "level" { + var err error + p.level, err = l.level() + if err != nil { + return p, err + } + } else if key == "usage_details" { err = l.usage(&p) } else if rowKeys[key] { var raw []byte diff --git a/internal/langfuse/level_test.go b/internal/langfuse/level_test.go new file mode 100644 index 0000000..2969f46 --- /dev/null +++ b/internal/langfuse/level_test.go @@ -0,0 +1,237 @@ +// SPDX-License-Identifier: Apache-2.0 +// Code authors: Vijay and Codex + +package langfuse + +import ( + "bytes" + "encoding/csv" + "encoding/json" + "fmt" + "strings" + "testing" +) + +func withLevel(row, raw string) string { + if raw == "" { + return row + } + return strings.TrimSuffix(strings.TrimSpace(row), "}") + `,"level":` + raw + "}\n" +} + +func levelEncoding(t *testing.T, rows []string, kind string) []byte { + t.Helper() + if kind == "JSONL" { + return []byte(strings.Join(rows, "")) + } + if kind == "JSON" { + return []byte("[" + strings.Join(strings.FieldsFunc(strings.TrimSpace(strings.Join(rows, "")), func(r rune) bool { return r == '\n' }), ",") + "]") + } + var b bytes.Buffer + w := csv.NewWriter(&b) + header := []string{"id", "project_id", "type", "start_time", "end_time", "provided_model_name", "user_id", "session_id", "environment", "release", "trace_name", "prompt_name", "tags", "usage_details", "total_cost", "level", "input", "output", "metadata", "tool_calls", "status_message"} + if err := w.Write(header); err != nil { + t.Fatal(err) + } + for _, row := range rows { + var fields map[string]json.RawMessage + if err := json.Unmarshal([]byte(row), &fields); err != nil { + t.Fatal(err) + } + values := make([]string, len(header)) + for i, key := range header { + if raw := fields[key]; raw != nil { + var s string + if json.Unmarshal(raw, &s) == nil { + values[i] = s + } else { + values[i] = string(raw) + } + } + } + if err := w.Write(values); err != nil { + t.Fatal(err) + } + } + w.Flush() + if err := w.Error(); err != nil { + t.Fatal(err) + } + return b.Bytes() +} + +func TestLevelEnumSixFormatsAndPrivacy(t *testing.T) { + raw := []string{`"DEFAULT"`, `"DEBUG"`, `"WARNING"`, `"ERROR"`, `"\u0045RROR"`, `"\u0057\u0041\u0052\u004e\u0049\u004e\u0047"`, "", `null`, `false`, `123`, `"error"`, `" ERROR"`, `"INFO"`, `"SUCCESS"`, `"PRIVATE_LEVEL"`, `{"private":"PRIVATE_LEVEL"}`, `"` + strings.Repeat("PRIVATE_LEVEL", 1000) + `"`, ` "ERROR"`} + want := []string{"DEFAULT", "DEBUG", "WARNING", "ERROR", "ERROR", "WARNING", "unknown", "unknown", "unknown", "unknown", "unknown", "unknown", "unknown", "unknown", "unknown", "unknown", "unknown", "ERROR"} + var rows []string + for i, level := range raw { + row := strings.TrimSpace(withLevel(rowJSON(fmt.Sprint(i), "project", "2026-10-10"), level)) + row = strings.TrimSuffix(row, "}") + `,"input":"PRIVATE_INPUT","output":{"x":"PRIVATE_OUTPUT"},"metadata":["PRIVATE_METADATA"],"tool_calls":"PRIVATE_TOOLS","status_message":"PRIVATE_STATUS"}` + "\n" + rows = append(rows, row) + } + var fingerprints []projectionFingerprint + for _, kind := range []string{"JSON", "JSONL", "CSV"} { + for _, zip := range []bool{false, true} { + for _, chunk := range []int{1, 7, 31} { + t.Run(fmt.Sprintf("%s/gzip=%v/chunk=%d", kind, zip, chunk), func(t *testing.T) { + data := levelEncoding(t, rows, kind) + if zip { + data = compressed(t, data) + } + i := 0 + report, err := readFile(chunkReader{bytes.NewReader(data), chunk}, descriptor(kind, zip, data, int64(len(rows))), func(p projection) error { + row, _, fingerprint, err := mappedRow(p, "user_id", "") + if err != nil || row.ObservationLevel != want[i] || row.Values["status"] != "" { + t.Fatal("enum projection or call-status boundary changed") + } + if kind == "JSON" && !zip && chunk == 1 { + fingerprints = append(fingerprints, fingerprint) + } else if fingerprints[i] != fingerprint { + t.Fatal("normalized duplicate fingerprint differs across formats") + } + for _, key := range []string{"level", "input", "output", "metadata", "tool_calls", "status_message"} { + if _, ok := p.fields[key]; ok { + t.Fatal("raw enum or forbidden payload retained") + } + } + encoded, _ := json.Marshal(row) + for _, sentinel := range []string{"PRIVATE_LEVEL", "PRIVATE_INPUT", "PRIVATE_OUTPUT", "PRIVATE_METADATA", "PRIVATE_TOOLS", "PRIVATE_STATUS"} { + if bytes.Contains(encoded, []byte(sentinel)) { + t.Fatal("private value leaked") + } + } + i++ + return nil + }) + if err != nil || i != len(rows) || report.InclusiveInput != uint64(len(rows)*100) || report.InclusiveOutput != uint64(len(rows)*40) || report.RecordedCost != "0.18" { + t.Fatal("level changed accounting or bounded reader", err) + } + }) + } + } + } +} + +func TestLevelEscapedKeyAndDuplicateCSVHeader(t *testing.T) { + data := []byte(strings.Replace(withLevel(rowJSON("one", "project", "2026-10-10"), `"ERROR"`), `"level"`, `"le\u0076el"`, 1)) + if err := readJSON(bytes.NewReader(data), false, func(p projection) error { + if p.level != "ERROR" { + t.Fatal("escaped level key was not projected") + } + return nil + }); err != nil { + t.Fatal(err) + } + data = []byte("id,level,level\none,ERROR,PRIVATE_LEVEL\n") + if err := readCSV(bytes.NewReader(data), func(projection) error { t.Fatal("duplicate header released a row"); return nil }); err == nil || strings.Contains(err.Error(), "PRIVATE_") { + t.Fatal("duplicate level header accepted or leaked") + } +} + +func TestLevelSeverityReviewAndErrorSubset(t *testing.T) { + root := t.TempDir() + var body string + for _, day := range []string{"2026-10-10", "2026-10-17"} { + for i, level := range []string{`"DEFAULT"`, `"DEBUG"`, `"WARNING"`, `"ERROR"`, "", `"PRIVATE_LEVEL"`} { + body += withLevel(rowJSON(fmt.Sprintf("%s-%d", day, i), "project", day), level) + } + } + addRun(t, root, "project", "run", "2026-10-10T00:00:00Z", "2026-10-18T00:00:00Z", body, 12) + r, err := Investigate(root, "user_id", "", options()) + if err != nil { + t.Fatal(err) + } + for _, w := range []struct { + tokens uint64 + cost string + unknown, unclear, failed, errors uint64 + }{ + {r.Before.Tokens, r.Before.RecordedSpend, r.Before.Quality.UnknownStatus, r.Before.Quality.Unclear, r.Before.Quality.Failed, r.Before.Quality.ErrorObservations}, + {r.After.Tokens, r.After.RecordedSpend, r.After.Quality.UnknownStatus, r.After.Quality.Unclear, r.After.Quality.Failed, r.After.Quality.ErrorObservations}, + } { + if w.tokens != 840 || w.cost != "0.06" || w.unknown != 6 || w.unclear != 3 || w.failed != 0 || w.errors != 1 { + t.Fatal("severity was treated as success/failure or changed totals/review") + } + } + if r.Before.ErrorObservationTokens == nil || r.Before.ErrorObservationTokens.Total != 140 || r.Before.ErrorObservationTokens.Input != 100 || r.Before.ErrorObservationTokens.Output != 40 || r.Before.ErrorObservationTokens.Finality != "unknown" || r.Before.FailureTokens != nil || r.Volume != nil || len(r.Models) != 1 || r.Models[0].Volume != nil || r.Models[0].Before.ErrorObservationTokens.Total != 140 { + t.Fatal("ERROR subset or conservative call-status gate differs") + } + for _, level := range []string{"DEFAULT", "DEBUG", "WARNING", "ERROR"} { + if r.Before.Quality.ObservationLevels[level] != 1 || r.After.Quality.ObservationLevels[level] != 1 { + t.Fatal("known severity counter differs") + } + } + if r.Before.Quality.ObservationLevels["unknown"] != 2 { + t.Fatal("unknown severity not counted") + } + b, _ := json.Marshal(r) + if bytes.Contains(b, []byte("PRIVATE_")) { + t.Fatal("private severity or identity leaked") + } +} + +func TestLevelDedupAndConflict(t *testing.T) { + for _, other := range []string{`"\u0045RROR"`, `"DEFAULT"`, "", `"PRIVATE_LEVEL"`} { + t.Run(other, func(t *testing.T) { + root := t.TempDir() + row := rowJSON("one", "project", "2026-10-10") + addRun(t, root, "project", "one", "2026-10-10T00:00:00Z", "2026-10-18T00:00:00Z", withLevel(row, `"ERROR"`), 1) + addRun(t, root, "project", "two", "2026-10-10T00:00:00Z", "2026-10-18T00:00:00Z", withLevel(row, other), 1) + r, err := Investigate(root, "user_id", "", options()) + if other == `"\u0045RROR"` { + if err != nil || r.Coverage.Duplicates != 1 || r.Before.Quality.ErrorObservations != 1 { + t.Fatal("equivalent enum not deduplicated", err) + } + } else if err == nil || !strings.Contains(err.Error(), "field: level") || strings.Contains(err.Error(), "PRIVATE_") || r.Before.Requests != 0 { + t.Fatal("conflict accepted, leaked or returned partial report") + } + }) + } +} + +func TestMalformedLevelDoesNotReleasePrefix(t *testing.T) { + for _, suffix := range []string{`,"level":"ERROR","le\u0076el":"DEFAULT"}`, `,"level":"\uD800"}`, `,"level":{"x":"PRIVATE_LEVEL"}`} { + row := strings.TrimSuffix(strings.TrimSpace(rowJSON("two", "project", "2026-10-17")), "}") + suffix + if strings.Contains(suffix, "PRIVATE_LEVEL") { + row = strings.TrimSuffix(row, "}") // malformed nested value + } + data := []byte(withLevel(rowJSON("one", "project", "2026-10-10"), `"DEFAULT"`) + row) + r, err := ReadFile(bytes.NewReader(data), descriptor("JSONL", false, data, 2)) + if err == nil || r.PhysicalRows != 0 || strings.Contains(err.Error(), "PRIVATE_") { + t.Fatal("malformed level released aggregate prefix or private diagnostics") + } + } +} + +func TestDuplicateConflictNamesFieldWithoutValues(t *testing.T) { + for _, tc := range []struct{ field, before, after string }{ + {"type", `"type":"GENERATION"`, `"type":"SPAN"`}, + {"start_time", "12:00:00.000000", "12:00:00.000001"}, + {"end_time", "12:00:01.000000", "12:00:02.000000"}, + {"total_cost", `"total_cost":0.01`, `"total_cost":0.02`}, + {"usage_details", `"input":60`, `"input":61`}, + {"provided_model_name", "PRIVATE_MODEL", "PRIVATE_MODEL_OTHER"}, + {"user_id", "PRIVATE_USER", "PRIVATE_USER_OTHER"}, + {"session_id", "PRIVATE_SESSION", "PRIVATE_SESSION_OTHER"}, + {"environment", "PRIVATE_ENV", "PRIVATE_ENV_OTHER"}, + {"release", "PRIVATE_RELEASE", "PRIVATE_RELEASE_OTHER"}, + {"trace_name", "PRIVATE_TRACE", "PRIVATE_TRACE_OTHER"}, + {"prompt_name", "PRIVATE_PROMPT", "PRIVATE_PROMPT_OTHER"}, + {"tags", "PRIVATE_TAG", "PRIVATE_TAG_OTHER"}, + } { + t.Run(tc.field, func(t *testing.T) { + root := t.TempDir() + before := rowJSON("one", "project", "2026-10-10") + after := strings.Replace(before, tc.before, tc.after, 1) + if before == after { + t.Fatal("conflict fixture did not change") + } + addRun(t, root, "project", "one", "2026-10-10T00:00:00Z", "2026-10-18T00:00:00Z", before, 1) + addRun(t, root, "project", "two", "2026-10-10T00:00:00Z", "2026-10-18T00:00:00Z", after, 1) + r, err := Investigate(root, "user_id", "", options()) + if err == nil || !strings.Contains(err.Error(), "field: "+tc.field) || strings.Contains(err.Error(), "PRIVATE_") || r.Before.Requests != 0 { + t.Fatal("conflict field absent, values leaked or partial report returned", err) + } + }) + } +} diff --git a/internal/langfuse/mapped_usage_test.go b/internal/langfuse/mapped_usage_test.go index f8ae39a..5a0ea99 100644 --- a/internal/langfuse/mapped_usage_test.go +++ b/internal/langfuse/mapped_usage_test.go @@ -13,10 +13,11 @@ import ( "github.com/llm-measurement/fleetdiff/internal/spend" ) -func mappedCapturedVariant(t *testing.T, details map[string]any) (spend.Row, [32]byte, [32]byte, error) { +func mappedCapturedVariant(t *testing.T, details map[string]any) (spend.Row, [32]byte, projectionFingerprint, error) { t.Helper() var row spend.Row - var id, fingerprint [32]byte + var id [32]byte + var fingerprint projectionFingerprint var mappedErr error err := readJSON(bytes.NewReader(capturedUsageVariant(t, details)), false, func(p projection) error { row, id, fingerprint, mappedErr = mappedRow(p, "user_id", "") diff --git a/internal/langfuse/testdata/fresh-r3/report.json b/internal/langfuse/testdata/fresh-r3/report.json index 630a1b3..0acf7aa 100644 --- a/internal/langfuse/testdata/fresh-r3/report.json +++ b/internal/langfuse/testdata/fresh-r3/report.json @@ -156,6 +156,9 @@ "zero_only_origin_unknown": 0, "failed_records": 0, "unknown_status": 5, + "observation_levels": { + "unknown": 5 + }, "unclear_usage_records": 5, "missing_group_identity": 0, "missing_recorded_spend": 0, @@ -219,6 +222,9 @@ "zero_only_origin_unknown": 0, "failed_records": 0, "unknown_status": 5, + "observation_levels": { + "unknown": 5 + }, "unclear_usage_records": 5, "missing_group_identity": 0, "missing_recorded_spend": 0, @@ -285,6 +291,9 @@ "zero_only_origin_unknown": 0, "failed_records": 0, "unknown_status": 5, + "observation_levels": { + "unknown": 5 + }, "unclear_usage_records": 5, "missing_group_identity": 0, "missing_recorded_spend": 0, @@ -345,6 +354,9 @@ "zero_only_origin_unknown": 0, "failed_records": 0, "unknown_status": 5, + "observation_levels": { + "unknown": 5 + }, "unclear_usage_records": 5, "missing_group_identity": 0, "missing_recorded_spend": 0, @@ -410,7 +422,8 @@ "Native input, cache-read and cache-creation are disjoint; inclusive input is their sum. Inclusive output includes disjoint reasoning.", "Recorded cost (Langfuse) is the exact sum of usable total_cost values, not an invoice. Zero cost with usage has unknown origin.", "Manifest coverage validates the supplied local copy, not upstream ingestion or hidden provider retries. Missing/partial files are not zero usage.", - "Raw user, session and custom labels are suppressed with local keyed aliases. Status is unknown; status_message is never decoded." + "Raw user, session and custom labels are suppressed with local keyed aliases. Status is unknown; status_message is never decoded.", + "Level is an observation severity enum, not provider success or billing finality. DEFAULT/DEBUG/WARNING alone do not require usage review; ERROR and missing/invalid levels do. Other usage checks still apply. Per-request splits remain unavailable while call status is unknown." ], "langfuse_coverage": { "manifests": 2, diff --git a/internal/spend/accounting.go b/internal/spend/accounting.go index f951fb8..003b581 100644 --- a/internal/spend/accounting.go +++ b/internal/spend/accounting.go @@ -53,18 +53,20 @@ func attributes(row Row) map[string]*commonpb.AnyValue { } type Quality struct { - Missing uint64 `json:"missing_token_fields"` - Invalid uint64 `json:"invalid_usage"` - TotalMismatch uint64 `json:"component_total_mismatch"` - TotalMissing uint64 `json:"missing_total_field"` - ZeroUnknown uint64 `json:"zero_only_origin_unknown"` - Failed uint64 `json:"failed_records"` - UnknownStatus uint64 `json:"unknown_status"` - Unclear uint64 `json:"unclear_usage_records"` - MissingIdentity uint64 `json:"missing_group_identity"` - SpendMissing uint64 `json:"missing_recorded_spend"` - SpendInvalid uint64 `json:"invalid_recorded_spend"` - SpendZeroUnknown uint64 `json:"zero_spend_origin_unknown"` + Missing uint64 `json:"missing_token_fields"` + Invalid uint64 `json:"invalid_usage"` + TotalMismatch uint64 `json:"component_total_mismatch"` + TotalMissing uint64 `json:"missing_total_field"` + ZeroUnknown uint64 `json:"zero_only_origin_unknown"` + Failed uint64 `json:"failed_records"` + UnknownStatus uint64 `json:"unknown_status"` + ObservationLevels map[string]uint64 `json:"observation_levels,omitempty"` + ErrorObservations uint64 `json:"error_observations,omitempty"` + Unclear uint64 `json:"unclear_usage_records"` + MissingIdentity uint64 `json:"missing_group_identity"` + SpendMissing uint64 `json:"missing_recorded_spend"` + SpendInvalid uint64 `json:"invalid_recorded_spend"` + SpendZeroUnknown uint64 `json:"zero_spend_origin_unknown"` } func classify(row Row, a accounting.Observation) Quality { @@ -95,7 +97,21 @@ func classify(row Row, a accounting.Observation) Quality { default: q.UnknownStatus = 1 } - if q.Missing+q.Invalid+q.TotalMismatch+q.TotalMissing+q.ZeroUnknown+q.Failed+q.UnknownStatus > 0 { + statusReview := q.UnknownStatus + if row.ObservationLevel != "" { + level := row.ObservationLevel + switch level { + case "DEFAULT", "DEBUG", "WARNING": + statusReview = 0 + case "ERROR": + q.ErrorObservations = 1 + default: + level = "unknown" + statusReview = 1 + } + q.ObservationLevels = map[string]uint64{level: 1} + } + if q.Missing+q.Invalid+q.TotalMismatch+q.TotalMissing+q.ZeroUnknown+q.Failed+statusReview+q.ErrorObservations > 0 { q.Unclear = 1 } return q @@ -109,6 +125,13 @@ func (q *Quality) add(v Quality) { q.ZeroUnknown += v.ZeroUnknown q.Failed += v.Failed q.UnknownStatus += v.UnknownStatus + q.ErrorObservations += v.ErrorObservations + if v.ObservationLevels != nil && q.ObservationLevels == nil { + q.ObservationLevels = map[string]uint64{} + } + for k, n := range v.ObservationLevels { + q.ObservationLevels[k] += n + } q.Unclear += v.Unclear q.MissingIdentity += v.MissingIdentity q.SpendMissing += v.SpendMissing diff --git a/internal/spend/coverage.go b/internal/spend/coverage.go index f6cdf41..46e6c9f 100644 --- a/internal/spend/coverage.go +++ b/internal/spend/coverage.go @@ -124,7 +124,7 @@ func orderCandidatesRead(path string, r *Report, key, expected []byte, selection digest := sha256.New() enc := json.NewEncoder(digest) err := read(path, func(row Row) error { - if err := enc.Encode(row.Values); err != nil { + if err := enc.Encode(row.replayProjection()); err != nil { return err } start, err := timestamp(row.Values["startTime"]) diff --git a/internal/spend/investigate.go b/internal/spend/investigate.go index d249d76..0a9ee76 100644 --- a/internal/spend/investigate.go +++ b/internal/spend/investigate.go @@ -74,11 +74,20 @@ func (a *aggregate) add(row Row, obs accounting.Observation, q Quality) error { return err } a.window.Tokens = total - if q.Failed != 0 && obs.Metrics[accounting.Prefix+"total_tokens_total"] != 0 { - if a.window.FailureTokens == nil { - a.window.FailureTokens = &FailureTokens{Finality: "unknown"} + for _, subset := range []struct { + count uint64 + dst **FailureTokens + }{ + {q.Failed, &a.window.FailureTokens}, + {q.ErrorObservations, &a.window.ErrorObservationTokens}, + } { + if subset.count == 0 || obs.Metrics[accounting.Prefix+"total_tokens_total"] == 0 { + continue + } + if *subset.dst == nil { + *subset.dst = &FailureTokens{Finality: "unknown"} } - f := a.window.FailureTokens + f := *subset.dst for _, field := range []struct { dst *uint64 value uint64 @@ -121,7 +130,7 @@ func (a *aggregate) result() Window { } func volume(a, b Window) *compare.VolumeChange { - if a.Quality.Unclear != 0 || b.Quality.Unclear != 0 { + if a.Quality.Unclear != 0 || b.Quality.Unclear != 0 || a.Quality.UnknownStatus != 0 || b.Quality.UnknownStatus != 0 { return nil } return compare.Decompose(a.Tokens, b.Tokens, a.Requests, b.Requests) @@ -226,7 +235,7 @@ func investigate(path string, opts Options, source func(string, func(Row) error) if latest.IsZero() || t.After(latest) { latest = t } - return enc.Encode(row.Values) + return enc.Encode(row.replayProjection()) }) if err != nil { return Report{}, err @@ -257,7 +266,7 @@ func investigate(path string, opts Options, source func(string, func(Row) error) enc := json.NewEncoder(secondDigest) err = read(path, func(row Row) error { r.Rows++ - if e := enc.Encode(row.Values); e != nil { + if e := enc.Encode(row.replayProjection()); e != nil { return e } if row.Values["request_id"] == "" { diff --git a/internal/spend/observation_level_test.go b/internal/spend/observation_level_test.go new file mode 100644 index 0000000..0a8a30d --- /dev/null +++ b/internal/spend/observation_level_test.go @@ -0,0 +1,96 @@ +// SPDX-License-Identifier: Apache-2.0 +// Code authors: Vijay and Codex + +package spend + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/llm-measurement/fleetdiff/internal/accounting" +) + +func TestObservationLevelPreservesAccountingFlags(t *testing.T) { + for _, level := range []string{"DEFAULT", "DEBUG", "WARNING", "ERROR", "unknown", "PRIVATE_LEVEL"} { + for _, mode := range []string{"clean", "missing", "invalid", "mismatch", "zero", "failure"} { + t.Run(level+"/"+mode, func(t *testing.T) { + row := Row{Values: sample("one", "2026-09-01", "key", 19, 3), ObservationLevel: level} + delete(row.Values, "status") + switch mode { + case "missing": + delete(row.Values, "completion_tokens") + case "invalid": + row.Values["prompt_tokens"] = "-1" + case "mismatch": + row.Values["total_tokens"] = "99" + case "zero": + row.Values["prompt_tokens"], row.Values["completion_tokens"], row.Values["total_tokens"] = "0", "0", "0" + case "failure": + row.Values["status"] = "failure" + } + obs, err := accounting.Tokens(attributes(row)) + if err != nil { + t.Fatal(err) + } + q := classify(row, obs) + wantReview := uint64(1) + if mode == "clean" && (level == "DEFAULT" || level == "DEBUG" || level == "WARNING") { + wantReview = 0 + } + if q.Unclear != wantReview || mode != "failure" && q.UnknownStatus != 1 || mode == "failure" && q.Failed != 1 { + t.Fatal("severity cleared accounting/status flag or implied success") + } + b, _ := json.Marshal(q) + if strings.Contains(string(b), "PRIVATE_") { + t.Fatal("unrecognized severity leaked") + } + }) + } + } +} + +func TestObservationLevelIsBoundToEveryReplay(t *testing.T) { + for _, automatic := range []bool{false, true} { + for _, changeOnPass := range []int{2, 3} { + opts := options(t) + if automatic { + opts.BeforePeriod, opts.AfterPeriod = "", "" + } + pass := 0 + _, err := InvestigateRows(func(visit func(Row) error) error { + pass++ + level := "DEFAULT" + if pass >= changeOnPass { + level = "ERROR" + } + for _, day := range []string{"2026-09-01", "2026-09-02", "2026-09-15", "2026-09-22"} { + row := Row{Values: sample(day, day, "key", 19, 3), ObservationLevel: level} + delete(row.Values, "status") + if e := visit(row); e != nil { + return e + } + } + return nil + }, opts) + // Explicit periods need two passes; automatic periods add a date pass. + if changeOnPass == 3 && !automatic { + if err != nil { + t.Fatal(err) + } + } else if err == nil || !strings.Contains(err.Error(), "changed") { + t.Fatal("changed severity escaped the completed-snapshot binding", err) + } + } + } +} + +func TestSQLCannotInjectObservationSeverity(t *testing.T) { + row := sample("one", "2026-09-01", "key", 19, 3) + delete(row, "status") + row["level"], row["ObservationLevel"], row["observation_level"] = "DEFAULT", "DEFAULT", "DEFAULT" + r, err := Investigate(writeRows(t, []map[string]string{row}), options(t)) + if err != nil || r.Before.Quality.UnknownStatus != 1 || r.Before.Quality.Unclear != 1 || r.Before.Quality.ObservationLevels != nil { + t.Fatal("SQL projection acquired adapter severity semantics", err) + } +} diff --git a/internal/spend/reader.go b/internal/spend/reader.go index ce77bc6..5f05eb9 100644 --- a/internal/spend/reader.go +++ b/internal/spend/reader.go @@ -37,6 +37,19 @@ const ( type Row struct { Values map[string]string Number int + // Set only by observation adapters, never by SQL/CSV status projections. + // Severity does not establish provider completion or billing finality. + ObservationLevel string +} + +func (r Row) replayProjection() any { + if r.ObservationLevel == "" { + return r.Values + } + return struct { + Values map[string]string + ObservationLevel string + }{r.Values, r.ObservationLevel} } var requiredColumns = [...]string{ diff --git a/internal/spend/report.go b/internal/spend/report.go index 356d120..ce14f7d 100644 --- a/internal/spend/report.go +++ b/internal/spend/report.go @@ -10,20 +10,21 @@ import ( ) type Window struct { - Cache compare.CacheWindow `json:"cache"` - Period Period `json:"period"` - Requests uint64 `json:"logged_model_requests"` - Tokens uint64 `json:"recorded_tokens"` - Counters map[string]uint64 `json:"counters"` - Quality Quality `json:"quality"` - FailureTokens *FailureTokens `json:"failure_tokens,omitempty"` - RecordedSpend string `json:"recorded_spend_usd"` - AttributedTokens uint64 `json:"attributed_tokens,omitempty"` - Distinct float64 `json:"distinct_groups_estimate,omitempty"` - DistinctNominalRSE float64 `json:"distinct_nominal_rse,omitempty"` + Cache compare.CacheWindow `json:"cache"` + Period Period `json:"period"` + Requests uint64 `json:"logged_model_requests"` + Tokens uint64 `json:"recorded_tokens"` + Counters map[string]uint64 `json:"counters"` + Quality Quality `json:"quality"` + FailureTokens *FailureTokens `json:"failure_tokens,omitempty"` + ErrorObservationTokens *FailureTokens `json:"error_observation_tokens,omitempty"` + RecordedSpend string `json:"recorded_spend_usd"` + AttributedTokens uint64 `json:"attributed_tokens,omitempty"` + Distinct float64 `json:"distinct_groups_estimate,omitempty"` + DistinctNominalRSE float64 `json:"distinct_nominal_rse,omitempty"` } -// These are the same gateway-recorded components included in recorded totals, +// These are the same source-recorded components included in recorded totals, // not independently confirmed provider receipts. type FailureTokens struct { Records uint64 `json:"records_with_tokens"`