fix(routing): 5 correctness bugs surfaced by adversarial code audit

database/migrations/20260911_030000_add_route_decisions_actual_upstream.sql

@@ -0,0 +1,5 @@
+-- Add actual_upstream_id to route_decisions so failover attempts can record
+-- which upstream actually served the successful attempt without discarding
+-- the initial pick's selected_upstream_id / candidate evaluations.
+ALTER TABLE route_decisions
+    ADD COLUMN IF NOT EXISTS actual_upstream_id BIGINT NOT NULL DEFAULT 0;

internal/forward/audit.go

@@ -247,9 +247,27 @@ func usageFromValue(value any) tokenUsage {
 	return out
 }
 
-// parseUsageObject 将 OpenAI 与 Claude 的字段名归一为统一计数。
+// parseUsageObject 将 OpenAI / Anthropic / Gemini 的 usage 归一为统一计数。
+//
+// 语义 (Anthropic 约定):
+//   input          = uncached prompt tokens (排他,不含 cached / cacheCreation)
+//   cached         = cache_read tokens (命中缓存部分)
+//   cacheCreation  = cache_write tokens (首次写入部分)
+//   output         = completion / response tokens
+//
+// 协议差异:
+//   Anthropic:  usage.input_tokens 本身就是 uncached; cache_read_input_tokens /
+//               cache_creation_input_tokens 独立字段。
+//   OpenAI:     usage.prompt_tokens 是 INCLUSIVE 的 —— 已经包含 cached。
+//               cached 出现在 usage.cached_tokens 或 prompt_tokens_details.cached_tokens。
+//   Gemini:     usage.promptTokenCount 是 INCLUSIVE 的 —— 已经包含 cachedContentTokenCount。
+//
+// 因此对 OpenAI/Gemini 需要在这里减去 cached,让存入 routing_observations 的
+// input 列在所有协议下保持"uncached"这个不变量。下游 SQL (CacheCoverageRatio,
+// TokenInflationFactor) 依赖这个不变量;不做归一它们会双数 cached。
 func parseUsageObject(usage map[string]any) tokenUsage {
-	input := maxInt64(maxInt64(number(usage["input_tokens"]), number(usage["prompt_tokens"])), number(usage["promptTokenCount"]))
+	inputAnthropic := number(usage["input_tokens"])
+	inputInclusive := maxInt64(number(usage["prompt_tokens"]), number(usage["promptTokenCount"]))
 	output := maxInt64(maxInt64(maxInt64(number(usage["output_tokens"]), number(usage["completion_tokens"])), number(usage["candidatesTokenCount"])), number(usage["thoughtsTokenCount"]))
 	cached := maxInt64(maxInt64(number(usage["cached_tokens"]), number(usage["cache_read_input_tokens"])), number(usage["cachedContentTokenCount"]))
 	cacheCreation := maxInt64(number(usage["cache_creation_tokens"]), number(usage["cache_creation_input_tokens"]))
@@ -258,6 +276,18 @@ func parseUsageObject(usage map[string]any) tokenUsage {
 			cached = maxInt64(cached, number(details["cached_tokens"]))
 		}
 	}
+	// Anthropic 已经是排他语义,直接采用其 input_tokens。
+	// OpenAI/Gemini 的 inputInclusive 减 cached 得到与 Anthropic 一致的排他值。
+	input := inputAnthropic
+	if inputInclusive > input {
+		normalized := inputInclusive - cached
+		if normalized < 0 {
+			normalized = 0
+		}
+		if normalized > input {
+			input = normalized
+		}
+	}
 	return tokenUsage{input: input, output: output, cached: cached, cacheCreation: cacheCreation}
 }
 

internal/forward/audit_test.go

@@ -35,6 +35,35 @@ func TestUsageAuditParsesAnthropicCacheTokens(t *testing.T) {
 	}
 }
 
+// OpenAI's prompt_tokens is inclusive of cached_tokens; Anthropic's input_tokens
+// is exclusive. Downstream SQL (CacheCoverageRatio, TokenInflationFactor)
+// assumes a single convention, so we normalize at parse time to Anthropic's
+// "uncached input" semantics.
+func TestUsageAuditNormalizesOpenAIPromptTokensToUncached(t *testing.T) {
+	// prompt_tokens=1000 includes cached=200 → uncached=800.
+	usage := usageFromJSON([]byte(`{"usage":{"prompt_tokens":1000,"completion_tokens":50,"prompt_tokens_details":{"cached_tokens":200}}}`))
+	if usage.input != 800 || usage.cached != 200 || usage.output != 50 {
+		t.Fatalf("openai normalization failed: %+v", usage)
+	}
+}
+
+// Gemini's promptTokenCount is inclusive of cachedContentTokenCount.
+func TestUsageAuditNormalizesGeminiPromptTokensToUncached(t *testing.T) {
+	usage := usageFromJSON([]byte(`{"usageMetadata":{"promptTokenCount":500,"candidatesTokenCount":40,"cachedContentTokenCount":150}}`))
+	if usage.input != 350 || usage.cached != 150 || usage.output != 40 {
+		t.Fatalf("gemini normalization failed: %+v", usage)
+	}
+}
+
+// Guard: if cached > prompt_tokens (garbage upstream), clamp to zero
+// instead of going negative.
+func TestUsageAuditClampsNegativeUncached(t *testing.T) {
+	usage := usageFromJSON([]byte(`{"usage":{"prompt_tokens":10,"cached_tokens":50}}`))
+	if usage.input != 0 || usage.cached != 50 {
+		t.Fatalf("negative uncached must clamp to zero: %+v", usage)
+	}
+}
+
 func TestContentBlockStopIsNotWholeStreamCompletion(t *testing.T) {
 	audit := &responseAudit{stream: true}
 	audit.feed([]byte("event: content_block_stop\ndata: {\"type\":\"content_block_stop\"}\n\n"))
@@ -58,7 +87,9 @@ func TestRelayResponseCapturesUsageBytesAndRequestID(t *testing.T) {
 	if result.err != nil || result.bytesSent != int64(len(body)) {
 		t.Fatalf("relay result mismatch: %+v", result)
 	}
-	if result.usage.input != 9 || result.usage.output != 4 || result.usage.cached != 2 {
+	// input is normalized to Anthropic-style "uncached" semantics: OpenAI
+	// prompt_tokens (9) is inclusive of cached (2), so uncached = 9 - 2 = 7.
+	if result.usage.input != 7 || result.usage.output != 4 || result.usage.cached != 2 {
 		t.Fatalf("usage mismatch: %+v", result.usage)
 	}
 	if result.upstreamRequestID != "upstream-123" {

internal/routing/selector.go

@@ -201,7 +201,7 @@ func Choose(request Request) (Decision, error) {
 		return a.CandidateID < b.CandidateID
 	})
 	winnerIndex := shortlist[0]
-	if explored, ok := chooseExploration(request, cfg, eligible, winnerIndex, now); ok {
+	if explored, ok := chooseExploration(request, cfg, decision.Evaluations, eligible, winnerIndex, now); ok {
 		winnerIndex = explored
 		decision.Exploration = true
 	}
@@ -250,7 +250,7 @@ func Choose(request Request) (Decision, error) {
 // still sending a small, bounded sample to less-observed eligible channels.
 // Only explores candidates within CostTieTolerance of the runner-up to avoid
 // wasting money on channels that are obviously more expensive.
-func chooseExploration(request Request, cfg Config, eligible []int, winner int, now time.Time) (int, bool) {
+func chooseExploration(request Request, cfg Config, evaluations []CandidateEvaluation, eligible []int, winner int, now time.Time) (int, bool) {
 	if request.Now.IsZero() || cfg.ExplorationRate <= 0 || len(eligible) < 2 {
 		return 0, false
 	}
@@ -260,26 +260,46 @@ func chooseExploration(request Request, cfg Config, eligible []int, winner int,
 	}
 	bucket := now.UnixNano() / bucketWindow.Nanoseconds()
 	hash := sha256.Sum256([]byte(request.Features.CacheKey + "\x00" + request.Features.Model + "\x00" + fmt.Sprint(bucket)))
-	threshold := uint64(cfg.ExplorationRate * float64(^uint64(0)))
+	// Rate >= 1 must always fire. float64 rounds ^uint64(0) up to 2^64, and
+	// the back-cast to uint64 is implementation-defined (on x86 it lands on
+	// 2^63), silently degrading rate=1.0 to ~50% fire rate.
+	var threshold uint64
+	if cfg.ExplorationRate >= 1 {
+		threshold = ^uint64(0)
+	} else {
+		threshold = uint64(cfg.ExplorationRate * float64(^uint64(0)))
+	}
 	if binary.BigEndian.Uint64(hash[:8]) > threshold {
 		return 0, false
 	}
-	// Only explore among candidates that are at most 2x the winner's cost.
-	// Expensive fallback channels are not worth exploring.
-	winnerCost := request.Candidates[winner].Price.Multiplier
-	if winnerCost <= 0 {
-		winnerCost = 1
+	// Only explore among candidates that are at most 2x the winner's forecast
+	// cost. The evaluation's EffectiveCost is the real per-decision price
+	// (SelectedTotal, possibly divided by success rate); Price.Multiplier is
+	// only a billing scalar and is typically ~1 across all channels, so it
+	// filters nothing in practice.
+	winnerCost := evaluations[winner].EffectiveCost
+	if math.IsNaN(winnerCost) || math.IsInf(winnerCost, 0) || winnerCost < 0 {
+		winnerCost = 0
 	}
+	// When winnerCost == 0 (e.g. the forecast reports zero requests or zero
+	// tokens), the 2x ceiling collapses to 0 and would filter every candidate.
+	// Fall back to a permissive ceiling in that case so exploration can still
+	// probe cold candidates during early warm-up.
 	costCeiling := winnerCost * 2
+	unbounded := winnerCost == 0
 	best := -1
 	for _, index := range eligible {
 		if index == winner {
 			continue
 		}
-		candidate := request.Candidates[index]
-		if candidate.Price.Multiplier > costCeiling {
+		cost := evaluations[index].EffectiveCost
+		if math.IsNaN(cost) || math.IsInf(cost, 0) || cost < 0 {
 			continue
 		}
+		if !unbounded && cost > costCeiling {
+			continue
+		}
+		candidate := request.Candidates[index]
 		if best < 0 || candidate.Performance.Samples < request.Candidates[best].Performance.Samples ||
 			(candidate.Performance.Samples == request.Candidates[best].Performance.Samples && candidate.ID < request.Candidates[best].ID) {
 			best = index

internal/routing/selector_test.go

@@ -2,6 +2,7 @@ package routing
 
 import (
 	"errors"
+	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -148,3 +149,60 @@ func TestChooseExploresLeastObservedEligibleCandidate(t *testing.T) {
 		t.Fatalf("expected exploration sample of cold candidate: %+v", decision)
 	}
 }
+
+// Exploration must NOT pick a candidate whose forecast cost is materially
+// higher than the winner's. Previously the ceiling compared Price.Multiplier
+// (typically ~1 everywhere), so a 100x-more-expensive fallback slipped through.
+func TestChooseExplorationRejectsExpensiveCandidates(t *testing.T) {
+	now := time.Unix(1_700_000_000, 0)
+	// Both eligible under identical multipliers, but "expensive" is 100x pricier
+	// per token; a real forecast must reject it as an exploration target.
+	cheap := healthyCandidate(1, "cheap", 0, basePrice(1e-7))
+	cheap.Performance = Performance{Samples: 100, SuccessRate: 1}
+	expensive := healthyCandidate(2, "expensive", 10, basePrice(1e-5))
+	expensive.Performance = Performance{Samples: 0}
+	cfg := DefaultConfig()
+	cfg.ExplorationRate = 1
+	// Give the request non-zero tokens so EffectiveCost > 0 and the ceiling
+	// is enforced. Otherwise every candidate has cost=0 and the safety
+	// fallback for winnerCost==0 kicks in.
+	decision, err := Choose(Request{
+		Features: RequestFeatures{
+			Model: "gpt-5", CacheKey: "session-expensive",
+			InputTokens: 10_000, EstimatedOutputTokens: 100,
+		},
+		Candidates: []Candidate{cheap, expensive}, Config: cfg, Now: now,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if decision.SelectedID == expensive.ID {
+		t.Fatalf("100x-more-expensive candidate must not be explored: %+v", decision)
+	}
+}
+
+// Rate=1.0 must always fire — regression for the float64→uint64 rounding bug
+// that silently degraded rate=1.0 to ~50% fire rate.
+func TestChooseExplorationRateOneAlwaysFires(t *testing.T) {
+	now := time.Unix(1_700_000_000, 0)
+	cold := healthyCandidate(2, "cold", 10, basePrice(1e-7))
+	cold.Performance.Samples = 0
+	warm := healthyCandidate(1, "warm", 0, basePrice(1e-7))
+	warm.Performance = Performance{Samples: 100, SuccessRate: 1}
+	cfg := DefaultConfig()
+	cfg.ExplorationRate = 1
+	// Try 20 distinct cache_keys — ALL should explore at rate=1.0 regardless
+	// of how the hash lands.
+	for i := 0; i < 20; i++ {
+		decision, err := Choose(Request{
+			Features:   RequestFeatures{Model: "gpt-5", CacheKey: fmt.Sprintf("s-%d", i)},
+			Candidates: []Candidate{warm, cold}, Config: cfg, Now: now,
+		})
+		if err != nil {
+			t.Fatalf("iter %d: %v", i, err)
+		}
+		if !decision.Exploration {
+			t.Fatalf("iter %d: exploration must fire at rate=1.0, got winner=%s", i, decision.SelectedName)
+		}
+	}
+}

internal/scheduler/intelligent.go

@@ -230,7 +230,7 @@ func (r *intelligentRouter) cache(item *upstream.Upstream, model string, feature
 		prefixHash = features.SessionID
 	}
 
-	stats, err := r.prefixStats(keyHash, item.ID, model, prefixHash, window, now)
+	stats, err := r.prefixStats(keyHash, item.ID, model, prefixHash, item.Protocol, window, now)
 	observed := err == nil
 	cacheObserved := observed && (stats.HitCount > 0 || stats.CreateCount > 0)
 	supported := cacheMode == upstream.CacheEnabled || cacheMode == upstream.CacheAuto || cacheObserved
@@ -384,15 +384,15 @@ func (r *intelligentRouter) upstreamStats(id int64, model string, window time.Du
 	return value, err
 }
 
-func (r *intelligentRouter) prefixStats(apiKeyHash string, upstreamID int64, model, prefixHash string, window time.Duration, now time.Time) (store.PrefixCacheStats, error) {
+func (r *intelligentRouter) prefixStats(apiKeyHash string, upstreamID int64, model, prefixHash, protocol string, window time.Duration, now time.Time) (store.PrefixCacheStats, error) {
 	key := prefixCacheKey{apiKeyHash: apiKeyHash, upstreamID: upstreamID, model: model, prefixHash: prefixHash, window: window}
 	r.mu.Lock()
 	if entry, ok := r.prefix[key]; ok && now.Before(entry.expires) {
 		r.mu.Unlock()
 		return entry.value, entry.err
 	}
 	r.mu.Unlock()
-	value, err := r.store.GetPrefixCacheStats(apiKeyHash, upstreamID, model, prefixHash, window, now)
+	value, err := r.store.GetPrefixCacheStats(apiKeyHash, upstreamID, model, prefixHash, protocol, window, now)
 	r.mu.Lock()
 	if r.prefix == nil {
 		r.prefix = make(map[prefixCacheKey]prefixCacheEntry)

internal/server/routing_audit.go

@@ -52,10 +52,17 @@ func (s *Server) persistRoutingAudit(requestID string, started time.Time, groupI
 			slog.Warn("save route decision failed", "request_id", requestID, "err", err)
 		} else {
 			_ = id
+			// ActualUpstreamID may differ from SelectedUpstreamID when the
+			// initial pick failed and a failover attempt succeeded. Persist
+			// it so downstream analysis can distinguish 'was picked' from
+			// 'actually served the successful attempt' — otherwise a JOIN by
+			// selected_upstream_id misattributes failover requests to the
+			// upstream that failed.
 			complete := store.RouteDecisionOutcome{
 				ActualInputTokens: optionalInt64Ptr(result.InputTokens), ActualOutputTokens: optionalInt64Ptr(result.OutputTokens),
 				ActualCachedTokens: optionalInt64Ptr(result.CachedTokens), ActualCacheCreationTokens: optionalInt64Ptr(result.CacheCreationTokens),
-				Outcome: result.Outcome, CompletedAt: time.Now(),
+				ActualUpstreamID: result.FinalUpstreamID,
+				Outcome:          result.Outcome, CompletedAt: time.Now(),
 			}
 			if err := s.store.CompleteRouteDecision(requestID, complete); err != nil {
 				slog.Warn("complete route decision failed", "request_id", requestID, "err", err)

internal/store/models_routing.go

@@ -35,6 +35,7 @@ type RouteDecisionModel struct {
 	ActualOutputTokens        *int64     `gorm:"column:actual_output_tokens;type:integer"`
 	ActualCachedTokens        *int64     `gorm:"column:actual_cached_tokens;type:integer"`
 	ActualCacheCreationTokens *int64     `gorm:"column:actual_cache_creation_tokens;type:integer"`
+	ActualUpstreamID          int64      `gorm:"column:actual_upstream_id;type:integer;not null;default:0"`
 	ActualOutcome             string     `gorm:"column:actual_outcome;type:text;not null;default:''"`
 	CreatedAt                 time.Time  `gorm:"column:created_at;not null;index:idx_route_decisions_created"`
 	CompletedAt               *time.Time `gorm:"column:completed_at"`

internal/store/route_decisions.go

@@ -77,8 +77,12 @@ type RouteDecisionOutcome struct {
 	ActualOutputTokens        *int64
 	ActualCachedTokens        *int64
 	ActualCacheCreationTokens *int64
-	Outcome                   string
-	CompletedAt               time.Time
+	// ActualUpstreamID is the upstream whose usage is being recorded. It may
+	// differ from RouteDecisionRecord.SelectedUpstreamID when the initial pick
+	// failed and a failover attempt succeeded. Zero means unknown / unchanged.
+	ActualUpstreamID int64
+	Outcome          string
+	CompletedAt      time.Time
 }
 
 type RouteDecisionEntry struct {
@@ -285,13 +289,19 @@ func (s *Store) CompleteRouteDecision(requestID string, outcome RouteDecisionOut
 	if outcome.CompletedAt.IsZero() {
 		outcome.CompletedAt = time.Now()
 	}
+	var actualUpstream *int64
+	if outcome.ActualUpstreamID > 0 {
+		actualUpstream = &outcome.ActualUpstreamID
+	}
 	result, err := s.exec(`UPDATE route_decisions SET
 		actual_cost=COALESCE(?,actual_cost),actual_input_tokens=COALESCE(?,actual_input_tokens),
 		actual_output_tokens=COALESCE(?,actual_output_tokens),actual_cached_tokens=COALESCE(?,actual_cached_tokens),
 		actual_cache_creation_tokens=COALESCE(?,actual_cache_creation_tokens),
+		actual_upstream_id=COALESCE(?,actual_upstream_id),
 		actual_outcome=CASE WHEN ?='' THEN actual_outcome ELSE ? END,completed_at=?
 		WHERE request_id=?`, outcome.ActualCost, outcome.ActualInputTokens, outcome.ActualOutputTokens,
-		outcome.ActualCachedTokens, outcome.ActualCacheCreationTokens, outcome.Outcome, outcome.Outcome,
+		outcome.ActualCachedTokens, outcome.ActualCacheCreationTokens, actualUpstream,
+		outcome.Outcome, outcome.Outcome,
 		s.timeValue(outcome.CompletedAt), requestID)
 	if err != nil {
 		return err