diff --git a/README.md b/README.md index 3f181271..d222db51 100644 --- a/README.md +++ b/README.md @@ -970,6 +970,15 @@ duplicate it. Attempt timings include any wait for a concurrency slot, while routing and retry delays remain visible as gaps inside the parent span. Tool execution inside the caller is outside the gateway's trace. +Model-call spans also send estimated USD input, output and total costs to +Langfuse via `langfuse.observation.cost_details`, using the same effective +prices as Magpie's usage ledger (custom prices first, then catalog prices). +Input cost includes cache reads and writes; reasoning tokens are already +included in output cost. Explicit zero prices are exported too. Unknown +prices are omitted so Langfuse can use its own model pricing. Parent and tool +spans carry no costs, avoiding duplicate counting. Subscription costs at +catalog prices are API-equivalent estimates, not subscription charges. + Enable **Trace agent conversations** to instead export one trace per user interaction, grouping model calls and tool executions under an agent root. Conversation IDs group those traces into Langfuse sessions. Gateway traces diff --git a/internal/usage/otel_cost_test.go b/internal/usage/otel_cost_test.go new file mode 100644 index 00000000..e19e7a61 --- /dev/null +++ b/internal/usage/otel_cost_test.go @@ -0,0 +1,142 @@ +package usage + +import ( + "encoding/json" + "math" + "os" + "path/filepath" + "testing" + "time" + + "github.com/yetone/magpie/internal/catalog" + "github.com/yetone/magpie/internal/settings" +) + +func TestOTelGenerationCosts(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + t.Setenv("XDG_CACHE_HOME", t.TempDir()) + if err := os.MkdirAll(filepath.Dir(catalog.CachePath()), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(catalog.CachePath(), []byte(`{"anthropic":{"id":"anthropic","models":{"claude-sonnet-5":{"id":"claude-sonnet-5","cost":{"input":3,"output":15,"cache_read":0.3,"cache_write":3.75}}}}}`), 0o600); err != nil { + t.Fatal(err) + } + catalog.Reset() + t.Cleanup(catalog.Reset) + price := func(in, out, read, write float64) settings.ModelPrice { + return settings.ModelPrice{Input: &in, Output: &out, CacheRead: &read, CacheWrite: &write} + } + config := settings.Settings{ModelPrices: map[string]settings.ModelPrice{ + "relay/claude-sonnet-5": price(2, 8, 0.2, 2.5), + "free/*": price(0, 0, 0, 0), + }} + if err := settings.Save(config); err != nil { + t.Fatal(err) + } + base := Record{Time: time.Now(), Provider: "relay", Model: "claude-sonnet-5", Input: 1000, Output: 200, CacheRead: 4000, CacheWrite: 500, Reasoning: 100, Status: 200} + exporter := &otelExporter{} + for _, tc := range []struct { + name, provider, model string + context *OTelSpan + status int + rejected, omit bool + input, output float64 + }{ + {name: "legacy generation", input: 0.00405, output: 0.0016}, + {name: "gateway attempt", context: &OTelSpan{}, input: 0.00405, output: 0.0016}, + {name: "billed failure", context: &OTelSpan{}, status: 502, input: 0.00405, output: 0.0016}, + {name: "local session generation", provider: UnknownProvider, context: &OTelSpan{Session: true, Type: "generation"}, input: 0.006075, output: 0.003}, + {name: "free model", provider: "free"}, + {name: "unknown price", model: "magpie-unpriced-test-model", omit: true}, + {name: "gateway parent", context: &OTelSpan{Root: true}, omit: true}, + {name: "conversation parent", context: &OTelSpan{Session: true, Type: "agent"}, omit: true}, + {name: "tool", context: &OTelSpan{Session: true, Type: "tool"}, omit: true}, + {name: "local rejection", rejected: true, omit: true}, + } { + t.Run(tc.name, func(t *testing.T) { + r := base + r.OTel, r.Rejected = tc.context, tc.rejected + if tc.provider != "" { + r.Provider = tc.provider + } + if tc.model != "" { + r.Model = tc.model + } + if tc.status != 0 { + r.Status = tc.status + } + // Decode the serialized OTLP envelope, including the nested JSON + // string Langfuse expects for costs, rather than testing helpers. + body, err := json.Marshal(exporter.traces([]Record{r})) + if err != nil { + t.Fatal(err) + } + var envelope struct { + Resources []struct { + Scopes []struct { + Spans []struct { + Attributes []otelAttribute `json:"attributes"` + } `json:"spans"` + } `json:"scopeSpans"` + } `json:"resourceSpans"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + t.Fatal(err) + } + var costs []string + for _, attr := range envelope.Resources[0].Scopes[0].Spans[0].Attributes { + if attr.Key == "langfuse.observation.cost_details" { + costs = append(costs, attr.Value["stringValue"].(string)) + } + } + if tc.omit { + if len(costs) != 0 { + t.Fatalf("unexpected costs: %v", costs) + } + return + } + if len(costs) != 1 { + t.Fatalf("want one cost attribute, got %v", costs) + } + var got map[string]float64 + if err := json.Unmarshal([]byte(costs[0]), &got); err != nil { + t.Fatal(err) + } + if len(got) != 3 { + t.Fatalf("cost breakdown: %v", got) + } + for key, want := range map[string]float64{"input": tc.input, "output": tc.output, "total": tc.input + tc.output} { + if value, ok := got[key]; !ok || math.Abs(value-want) > 1e-12 { + t.Errorf("%s cost = %v, want %v", key, value, want) + } + } + }) + } + // A later batch must pick up changed settings rather than retain tariffs + // from the exporter's first batch. + config.ModelPrices["relay/claude-sonnet-5"] = price(0, 0, 0, 0) + if err := settings.Save(config); err != nil { + t.Fatal(err) + } + body, err := json.Marshal(exporter.traces([]Record{base})) + if err != nil { + t.Fatal(err) + } + var got map[string]any + if err := json.Unmarshal(body, &got); err != nil { + t.Fatal(err) + } + resources := got["resourceSpans"].([]any) + scopes := resources[0].(map[string]any)["scopeSpans"].([]any) + spans := scopes[0].(map[string]any)["spans"].([]any) + for _, a := range spans[0].(map[string]any)["attributes"].([]any) { + attr := a.(map[string]any) + if attr["key"] == "langfuse.observation.cost_details" { + if attr["value"].(map[string]any)["stringValue"] != `{"input":0,"output":0,"total":0}` { + t.Fatalf("stale tariff: %v", attr) + } + return + } + } + t.Fatal("missing updated costs") +} diff --git a/internal/usage/otel_wire.go b/internal/usage/otel_wire.go index a03c5b24..68beb67a 100644 --- a/internal/usage/otel_wire.go +++ b/internal/usage/otel_wire.go @@ -64,6 +64,7 @@ type OTelSpan struct { func (e *otelExporter) traces(records []Record) any { spans := make([]any, 0, len(records)) + priceOf := pricer() for _, r := range records { var traceID [16]byte if r.OTel == nil { @@ -163,6 +164,24 @@ func (e *otelExporter) traces(records []Record) any { } span["attributes"] = a } + // Price only model calls, never their gateway/conversation parents or + // tools. Cache reads/writes belong to input cost; reasoning is already + // included in output tokens. Use the ledger's effective tariff, including + // explicit free models, and leave unknown prices for Langfuse to infer. + generation := r.OTel == nil || !r.OTel.Root && (!r.OTel.Session || r.OTel.Type == "generation") + if generation && !r.IsRejected() { + if pr := priceOf(r); pr != nil { + cost, err := json.Marshal(map[string]float64{ + "input": pr.Cost(r.Input, 0, r.CacheRead, r.CacheWrite), + "output": pr.Cost(0, r.Output, 0, 0), + "total": pr.Cost(r.Input, r.Output, r.CacheRead, r.CacheWrite), + }) + if err == nil { + a = append(a, otelString("langfuse.observation.cost_details", string(cost))) + span["attributes"] = a + } + } + } spans = append(spans, span) } return otelEnvelope("Spans", spans)