Merge branch 'fix/cache-hit-ratio-anthropic'

commit: a43179a5dbd46173696cd2c5ea8d8c6d2a5d1ce8
author: brice · 2d ago · 2026-10-20 05:44 UTC
merge of: 6bab2c0 + bcb8569 · diff against the first parent
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
checks: 0/3

2 files · +107 -16

M packages/provider/anthropic.go +34 -16


@@ -666,8 +666,12 @@ func (c anthropicClient) runStream(ctx context.Context, resp http.Response, re
666	666		blockOrder []int // insertion order of indexes
667	667		activeIdx = -1
668	668		usage Usage
669		-	stop = StopEnd
670		-	finalErr error
	669	+	// cacheReported is set once the stream carries explicit cache
	670	+	// usage fields, so the cache hit ratio is only reported when the
	671	+	// endpoint measured it (some Anthropic-compatible APIs omit them).
	672	+	cacheReported bool
	673	+	stop = StopEnd
	674	+	finalErr error
671	675		)
672	676		_ = activeIdx // read-only indicator used for legacy parity
673	677

@@ -700,6 +704,10 @@ func (c anthropicClient) runStream(ctx context.Context, resp http.Response, re
700	704		}
701	705
702	706		sendDone := func() {
	707	+	if cacheReported {
	708	+	usage.CacheMeasuredPromptTokens = usage.PromptTokens()
	709	+	usage.CacheMeasuredReadTokens = usage.CacheReadTokens
	710	+	}
703	711		usage.CostUSD = ComputeCost(model, usage)
704	712		out <- EventUsage{Usage: usage}
705	713		out <- EventDone{Stop: stop, Err: finalErr, Message: assembleMsg()}

@@ -800,10 +808,10 @@ func (c anthropicClient) runStream(ctx context.Context, resp http.Response, re
800	808		var m struct {
801	809		Message struct {
802	810		Usage struct {
803		-	InputTokens int `json:"input_tokens"`
804		-	OutputTokens int `json:"output_tokens"`
805		-	CacheReadInputTokens int `json:"cache_read_input_tokens"`
806		-	CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
	811	+	InputTokens int `json:"input_tokens"`
	812	+	OutputTokens int `json:"output_tokens"`
	813	+	CacheReadInputTokens *int `json:"cache_read_input_tokens"`
	814	+	CacheCreationInputTokens *int `json:"cache_creation_input_tokens"`
807	815		} `json:"usage"`
808	816		} `json:"message"`
809	817		}

@@ -814,18 +822,26 @@ func (c anthropicClient) runStream(ctx context.Context, resp http.Response, re
814	822		// which can be 50-70% of cost.
815	823		usage.InputTokens = m.Message.Usage.InputTokens
816	824		usage.OutputTokens = m.Message.Usage.OutputTokens
817		-	usage.CacheReadTokens = m.Message.Usage.CacheReadInputTokens
818		-	usage.CacheWriteTokens = m.Message.Usage.CacheCreationInputTokens
	825	+	usage.CacheReadTokens = 0
	826	+	usage.CacheWriteTokens = 0
	827	+	if r := m.Message.Usage.CacheReadInputTokens; r != nil {
	828	+	usage.CacheReadTokens = *r
	829	+	cacheReported = true
	830	+	}
	831	+	if w := m.Message.Usage.CacheCreationInputTokens; w != nil {
	832	+	usage.CacheWriteTokens = *w
	833	+	cacheReported = true
	834	+	}
819	835		case "message_delta":
820	836		var m struct {
821	837		Delta struct {
822	838		StopReason string `json:"stop_reason"`
823	839		} `json:"delta"`
824	840		Usage struct {
825		-	InputTokens int `json:"input_tokens"`
826		-	OutputTokens int `json:"output_tokens"`
827		-	CacheReadInputTokens int `json:"cache_read_input_tokens"`
828		-	CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
	841	+	InputTokens int `json:"input_tokens"`
	842	+	OutputTokens int `json:"output_tokens"`
	843	+	CacheReadInputTokens *int `json:"cache_read_input_tokens"`
	844	+	CacheCreationInputTokens *int `json:"cache_creation_input_tokens"`
829	845		} `json:"usage"`
830	846		}
831	847		_ = json.Unmarshal([]byte(ev.Data), &m)

@@ -838,11 +854,13 @@ func (c anthropicClient) runStream(ctx context.Context, resp http.Response, re
838	854		if m.Usage.OutputTokens > 0 {
839	855		usage.OutputTokens = m.Usage.OutputTokens
840	856		}
841		-	if m.Usage.CacheReadInputTokens > 0 {
842		-	usage.CacheReadTokens = m.Usage.CacheReadInputTokens
	857	+	if r := m.Usage.CacheReadInputTokens; r != nil {
	858	+	usage.CacheReadTokens = *r
	859	+	cacheReported = true
843	860		}
844		-	if m.Usage.CacheCreationInputTokens > 0 {
845		-	usage.CacheWriteTokens = m.Usage.CacheCreationInputTokens
	861	+	if w := m.Usage.CacheCreationInputTokens; w != nil {
	862	+	usage.CacheWriteTokens = *w
	863	+	cacheReported = true
846	864		}
847	865		switch m.Delta.StopReason {
848	866		case "end_turn", "stop_sequence":

 		blockOrder []int // insertion order of indexes
 		activeIdx  = -1
 		usage      Usage
-		stop       = StopEnd
+		// cacheReported is set once the stream carries explicit cache
-		finalErr   error
+		// usage fields, so the cache hit ratio is only reported when the
+		// endpoint measured it (some Anthropic-compatible APIs omit them).
+		cacheReported bool
+		stop          = StopEnd
+		finalErr      error
 	)
 	_ = activeIdx // read-only indicator used for legacy parity
 	}
 	sendDone := func() {
+		if cacheReported {
+			usage.CacheMeasuredPromptTokens = usage.PromptTokens()
+			usage.CacheMeasuredReadTokens = usage.CacheReadTokens
+		}
 		usage.CostUSD = ComputeCost(model, usage)
 		out <- EventUsage{Usage: usage}
 		out <- EventDone{Stop: stop, Err: finalErr, Message: assembleMsg()}
 				var m struct {
 					Message struct {
 						Usage struct {
-							InputTokens              int `json:"input_tokens"`
+							InputTokens              int  `json:"input_tokens"`
-							OutputTokens             int `json:"output_tokens"`
+							OutputTokens             int  `json:"output_tokens"`
-							CacheReadInputTokens     int `json:"cache_read_input_tokens"`
+							CacheReadInputTokens     *int `json:"cache_read_input_tokens"`
-							CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
+							CacheCreationInputTokens *int `json:"cache_creation_input_tokens"`
 						} `json:"usage"`
 					} `json:"message"`
 				}
 				// which can be 50-70% of cost.
 				usage.InputTokens = m.Message.Usage.InputTokens
 				usage.OutputTokens = m.Message.Usage.OutputTokens
-				usage.CacheReadTokens = m.Message.Usage.CacheReadInputTokens
+				usage.CacheReadTokens = 0
-				usage.CacheWriteTokens = m.Message.Usage.CacheCreationInputTokens
+				usage.CacheWriteTokens = 0
+				if r := m.Message.Usage.CacheReadInputTokens; r != nil {
+					usage.CacheReadTokens = *r
+					cacheReported = true
+				}
+				if w := m.Message.Usage.CacheCreationInputTokens; w != nil {
+					usage.CacheWriteTokens = *w
+					cacheReported = true
+				}
 			case "message_delta":
 				var m struct {
 					Delta struct {
 						StopReason string `json:"stop_reason"`
 					} `json:"delta"`
 					Usage struct {
-						InputTokens              int `json:"input_tokens"`
+						InputTokens              int  `json:"input_tokens"`
-						OutputTokens             int `json:"output_tokens"`
+						OutputTokens             int  `json:"output_tokens"`
-						CacheReadInputTokens     int `json:"cache_read_input_tokens"`
+						CacheReadInputTokens     *int `json:"cache_read_input_tokens"`
-						CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
+						CacheCreationInputTokens *int `json:"cache_creation_input_tokens"`
 					} `json:"usage"`
 				}
 				_ = json.Unmarshal([]byte(ev.Data), &m)
 				if m.Usage.OutputTokens > 0 {
 					usage.OutputTokens = m.Usage.OutputTokens
 				}
-				if m.Usage.CacheReadInputTokens > 0 {
+				if r := m.Usage.CacheReadInputTokens; r != nil {
-					usage.CacheReadTokens = m.Usage.CacheReadInputTokens
+					usage.CacheReadTokens = *r
+					cacheReported = true
 				}
-				if m.Usage.CacheCreationInputTokens > 0 {
+				if w := m.Usage.CacheCreationInputTokens; w != nil {
-					usage.CacheWriteTokens = m.Usage.CacheCreationInputTokens
+					usage.CacheWriteTokens = *w
+					cacheReported = true
 				}
 				switch m.Delta.StopReason {
 				case "end_turn", "stop_sequence":

M packages/provider/provider_test.go +73 -0


@@ -554,6 +554,79 @@ func TestAnthropicStreamHappyPath(t *testing.T) {
554	554		}
555	555		}
556	556
	557	+	func anthropicStreamUsage(t *testing.T, messageStart string) Usage {
	558	+	t.Helper()
	559	+	return anthropicStreamUsageWithDelta(t, messageStart, `{"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}`)
	560	+	}
	561	+
	562	+	func anthropicStreamUsageWithDelta(t *testing.T, messageStart, messageDelta string) Usage {
	563	+	t.Helper()
	564	+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	565	+	w.Header().Set("content-type", "text/event-stream")
	566	+	w.WriteHeader(http.StatusOK)
	567	+	_, _ = w.Write([]byte("event: message_start\ndata: " + messageStart + "\n\n" +
	568	+	"event: message_delta\ndata: " + messageDelta + "\n\n" +
	569	+	"event: message_stop\ndata: {}\n\n"))
	570	+	}))
	571	+	defer srv.Close()
	572	+
	573	+	evs, err := NewAnthropic("x", srv.URL).Stream(context.Background(), Request{Model: "claude-sonnet-4-5"})
	574	+	if err != nil {
	575	+	t.Fatal(err)
	576	+	}
	577	+	var usage Usage
	578	+	for ev := range evs {
	579	+	if e, ok := ev.(EventUsage); ok {
	580	+	usage = e.Usage
	581	+	}
	582	+	}
	583	+	return usage
	584	+	}
	585	+
	586	+	func TestAnthropicStreamReportsCacheHitRatio(t *testing.T) {
	587	+	usage := anthropicStreamUsage(t, `{"message":{"usage":{"input_tokens":20,"output_tokens":0,"cache_read_input_tokens":900,"cache_creation_input_tokens":80}}}`)
	588	+	if usage.CacheReadTokens != 900 \|\| usage.CacheWriteTokens != 80 {
	589	+	t.Fatalf("cache tokens = R%d W%d, want R900 W80", usage.CacheReadTokens, usage.CacheWriteTokens)
	590	+	}
	591	+	ratio, ok := usage.CacheHitRatio()
	592	+	if !ok \|\| math.Abs(ratio-0.9) > 1e-9 {
	593	+	t.Fatalf("cache hit ratio = %v, %v, want 0.9, true", ratio, ok)
	594	+	}
	595	+	}
	596	+
	597	+	func TestAnthropicStreamKnownCacheMissReportsZeroRatio(t *testing.T) {
	598	+	usage := anthropicStreamUsage(t, `{"message":{"usage":{"input_tokens":100,"output_tokens":0,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}}`)
	599	+	if ratio, ok := usage.CacheHitRatio(); !ok \|\| ratio != 0 {
	600	+	t.Fatalf("cache hit ratio = %v, %v, want 0, true", ratio, ok)
	601	+	}
	602	+	}
	603	+
	604	+	func TestAnthropicStreamCacheFieldsInDeltaOnly(t *testing.T) {
	605	+	start := `{"message":{"usage":{"input_tokens":100,"output_tokens":0}}}`
	606	+	for _, tc := range []struct {
	607	+	name string
	608	+	delta string
	609	+	want float64
	610	+	}{
	611	+	{"known miss", `{"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}`, 0},
	612	+	{"hit", `{"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5,"cache_read_input_tokens":300,"cache_creation_input_tokens":0}}`, 0.75},
	613	+	} {
	614	+	t.Run(tc.name, func(t *testing.T) {
	615	+	usage := anthropicStreamUsageWithDelta(t, start, tc.delta)
	616	+	if ratio, ok := usage.CacheHitRatio(); !ok \|\| math.Abs(ratio-tc.want) > 1e-9 {
	617	+	t.Fatalf("cache hit ratio = %v, %v, want %v, true", ratio, ok, tc.want)
	618	+	}
	619	+	})
	620	+	}
	621	+	}
	622	+
	623	+	func TestAnthropicStreamWithoutCacheFieldsLeavesRatioUnknown(t *testing.T) {
	624	+	usage := anthropicStreamUsage(t, `{"message":{"usage":{"input_tokens":100,"output_tokens":0}}}`)
	625	+	if ratio, ok := usage.CacheHitRatio(); ok {
	626	+	t.Fatalf("cache hit ratio = %v, true, want unknown", ratio)
	627	+	}
	628	+	}
	629	+
557	630		func TestClaudeOpus5Catalog(t *testing.T) {
558	631		m, err := FindModel("anthropic", "claude-opus-5")
559	632		if err != nil {


@@ -554,6 +554,79 @@ func TestAnthropicStreamHappyPath(t *testing.T) {
554	}	554		}
555	}	555		}
556		556
		557	+	func anthropicStreamUsage(t *testing.T, messageStart string) Usage {
		558	+	t.Helper()
		559	+	return anthropicStreamUsageWithDelta(t, messageStart, `{"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}`)
		560	+	}
		561	+
		562	+	func anthropicStreamUsageWithDelta(t *testing.T, messageStart, messageDelta string) Usage {
		563	+	t.Helper()
		564	+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		565	+	w.Header().Set("content-type", "text/event-stream")
		566	+	w.WriteHeader(http.StatusOK)
		567	+	_, _ = w.Write([]byte("event: message_start\ndata: " + messageStart + "\n\n" +
		568	+	"event: message_delta\ndata: " + messageDelta + "\n\n" +
		569	+	"event: message_stop\ndata: {}\n\n"))
		570	+	}))
		571	+	defer srv.Close()
		572	+
		573	+	evs, err := NewAnthropic("x", srv.URL).Stream(context.Background(), Request{Model: "claude-sonnet-4-5"})
		574	+	if err != nil {
		575	+	t.Fatal(err)
		576	+	}
		577	+	var usage Usage
		578	+	for ev := range evs {
		579	+	if e, ok := ev.(EventUsage); ok {
		580	+	usage = e.Usage
		581	+	}
		582	+	}
		583	+	return usage
		584	+	}
		585	+
		586	+	func TestAnthropicStreamReportsCacheHitRatio(t *testing.T) {
		587	+	usage := anthropicStreamUsage(t, `{"message":{"usage":{"input_tokens":20,"output_tokens":0,"cache_read_input_tokens":900,"cache_creation_input_tokens":80}}}`)
		588	+	if usage.CacheReadTokens != 900 \|\| usage.CacheWriteTokens != 80 {
		589	+	t.Fatalf("cache tokens = R%d W%d, want R900 W80", usage.CacheReadTokens, usage.CacheWriteTokens)
		590	+	}
		591	+	ratio, ok := usage.CacheHitRatio()
		592	+	if !ok \|\| math.Abs(ratio-0.9) > 1e-9 {
		593	+	t.Fatalf("cache hit ratio = %v, %v, want 0.9, true", ratio, ok)
		594	+	}
		595	+	}
		596	+
		597	+	func TestAnthropicStreamKnownCacheMissReportsZeroRatio(t *testing.T) {
		598	+	usage := anthropicStreamUsage(t, `{"message":{"usage":{"input_tokens":100,"output_tokens":0,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}}`)
		599	+	if ratio, ok := usage.CacheHitRatio(); !ok \|\| ratio != 0 {
		600	+	t.Fatalf("cache hit ratio = %v, %v, want 0, true", ratio, ok)
		601	+	}
		602	+	}
		603	+
		604	+	func TestAnthropicStreamCacheFieldsInDeltaOnly(t *testing.T) {
		605	+	start := `{"message":{"usage":{"input_tokens":100,"output_tokens":0}}}`
		606	+	for _, tc := range []struct {
		607	+	name string
		608	+	delta string
		609	+	want float64
		610	+	}{
		611	+	{"known miss", `{"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}`, 0},
		612	+	{"hit", `{"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5,"cache_read_input_tokens":300,"cache_creation_input_tokens":0}}`, 0.75},
		613	+	} {
		614	+	t.Run(tc.name, func(t *testing.T) {
		615	+	usage := anthropicStreamUsageWithDelta(t, start, tc.delta)
		616	+	if ratio, ok := usage.CacheHitRatio(); !ok \|\| math.Abs(ratio-tc.want) > 1e-9 {
		617	+	t.Fatalf("cache hit ratio = %v, %v, want %v, true", ratio, ok, tc.want)
		618	+	}
		619	+	})
		620	+	}
		621	+	}
		622	+
		623	+	func TestAnthropicStreamWithoutCacheFieldsLeavesRatioUnknown(t *testing.T) {
		624	+	usage := anthropicStreamUsage(t, `{"message":{"usage":{"input_tokens":100,"output_tokens":0}}}`)
		625	+	if ratio, ok := usage.CacheHitRatio(); ok {
		626	+	t.Fatalf("cache hit ratio = %v, true, want unknown", ratio)
		627	+	}
		628	+	}
		629	+
557	func TestClaudeOpus5Catalog(t *testing.T) {	630		func TestClaudeOpus5Catalog(t *testing.T) {
558	m, err := FindModel("anthropic", "claude-opus-5")	631		m, err := FindModel("anthropic", "claude-opus-5")
559	if err != nil {	632		if err != nil {