usage: export effective model costs to Langfuse (#684)

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

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")
+}

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)