From 79760b9bc8d113d7998d5c0e352d6888841ccadd Mon Sep 17 00:00:00 2001 From: Steven Moder Date: Fri, 11 Sep 2026 10:06:38 +0800 Subject: [PATCH 1/5] fix(routing): exploration ceiling uses EffectiveCost + rate=1.0 always fires MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two audit findings, one function: 1. (#6) chooseExploration compared Price.Multiplier * 2 as the cost ceiling, but Multiplier is a billing scalar that is ~1 across all channels in production, so the guard filtered nothing. A 100x-more-expensive fallback could receive exploration traffic silently. Fix: use the evaluation's EffectiveCost (SelectedTotal / success_rate) so the '2x winner' filter means what the comment says. 2. (#7) threshold := uint64(rate * float64(^uint64(0))) — float64 rounds ^uint64(0) up to 2^64, and casting 2^64 back to uint64 is implementation- defined (x86 lands on 2^63), degrading rate=1.0 to ~50% fire rate. Fix: short-circuit rate>=1 to always fire. Also added a permissive fallback when winnerCost==0 (zero-token forecast, early warm-up) so the 2x ceiling doesn't collapse to zero and forbid all exploration. Tests: - TestChooseExplorationRejectsExpensiveCandidates: 100x-priced fallback must never win at rate=1.0. - TestChooseExplorationRateOneAlwaysFires: 20 distinct cache_keys must ALL explore at rate=1.0. --- internal/routing/selector.go | 40 +++++++++++++++------ internal/routing/selector_test.go | 58 +++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 10 deletions(-) diff --git a/internal/routing/selector.go b/internal/routing/selector.go index 08d2d17..fe40e69 100644 --- a/internal/routing/selector.go +++ b/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 diff --git a/internal/routing/selector_test.go b/internal/routing/selector_test.go index 0d1b815..7566117 100644 --- a/internal/routing/selector_test.go +++ b/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) + } + } +} From 1ca15d864aa0e414dcaf4a9a69c5b710630a7dd4 Mon Sep 17 00:00:00 2001 From: Steven Moder Date: Fri, 11 Sep 2026 10:10:03 +0800 Subject: [PATCH 2/5] fix(store): align session-cache TTL fallback with selectAdaptiveTTL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit getSessionCacheStats stamped stats.ExpiresAt = latestCache + 5min unless the session had run >10min AND had CreateCount>=2. That missed two branches from routing.selectAdaptiveTTL: Gemini (always 1h) and any protocol with sparse avgInterval>4min+CreateCount>=1. In those cases the state machine at cachestate.go:131 flipped CacheHot → CacheExpired up to 55 minutes early: profile.Existing.Valid went false, cacheLifetimes lost the guaranteed initial hit credit, and the candidate over-forecasted a full extra cache write. Fix: extract the TTL policy into store.AssumedCacheTTL and call it from getSessionCacheStats. Same rules, single source of truth, kept in the store package to avoid a store → routing import cycle (routing.selectAdaptiveTTL retains its own copy — see doc-comment cross-reference). Signature change: GetPrefixCacheStats / getSessionCacheStats now take a protocol string. Callers updated (scheduler + tests). Fixes audit finding #1. --- internal/scheduler/intelligent.go | 6 +-- internal/store/routing_observations.go | 48 +++++++++++++++++----- internal/store/routing_persistence_test.go | 34 ++++++++++++++- 3 files changed, 72 insertions(+), 16 deletions(-) diff --git a/internal/scheduler/intelligent.go b/internal/scheduler/intelligent.go index b74a72f..e2c33a2 100644 --- a/internal/scheduler/intelligent.go +++ b/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,7 +384,7 @@ 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) { @@ -392,7 +392,7 @@ func (r *intelligentRouter) prefixStats(apiKeyHash string, upstreamID int64, mod 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) diff --git a/internal/store/routing_observations.go b/internal/store/routing_observations.go index 849482a..bac22cb 100644 --- a/internal/store/routing_observations.go +++ b/internal/store/routing_observations.go @@ -481,11 +481,40 @@ func (s *Store) GetUpstreamRoutingStats(upstreamID int64, model string, window t return stats, nil } +// AssumedCacheTTL picks the TTL to attribute to the most recent cache create +// when we only know aggregate observations for a session (no explicit ExpiresAt +// row). It mirrors the routing.selectAdaptiveTTL policy so the two encodings of +// the same rule stay in sync: Gemini → 1h; long session with rebuilds → 1h; +// sparse conversation with any rebuild → 1h; otherwise 5min default. +// +// Kept here (rather than importing routing) to avoid a store → routing cycle; +// see the routing.selectAdaptiveTTL doc-comment for the source of truth. +func AssumedCacheTTL(protocol string, observations, createCount int64, firstSeenAt, now int64) time.Duration { + const defaultTTL = 5 * time.Minute + const extendedTTL = time.Hour + switch strings.ToLower(strings.TrimSpace(protocol)) { + case "gemini", "google", "generativelanguage", "generatecontent": + return extendedTTL + } + if firstSeenAt <= 0 || observations <= 0 { + return defaultTTL + } + sessionDuration := time.Duration(now-firstSeenAt) * time.Second + if sessionDuration > 10*time.Minute && createCount >= 2 { + return extendedTTL + } + avgInterval := sessionDuration / time.Duration(observations) + if avgInterval > 4*time.Minute && createCount >= 1 { + return extendedTTL + } + return defaultTTL +} + // GetPrefixCacheStats reads the cache state isolated by upstream credential, // upstream, model, and prefix. Lifetime counters are returned alongside a // recent-window hit rate; zero/unknown expiry is treated conservatively as // not currently valid. -func (s *Store) GetPrefixCacheStats(apiKeyHash string, upstreamID int64, model, prefixHash string, window time.Duration, now time.Time) (PrefixCacheStats, error) { +func (s *Store) GetPrefixCacheStats(apiKeyHash string, upstreamID int64, model, prefixHash, protocol string, window time.Duration, now time.Time) (PrefixCacheStats, error) { if window <= 0 { window = DefaultRoutingStatsWindow } @@ -506,7 +535,7 @@ func (s *Store) GetPrefixCacheStats(apiKeyHash string, upstreamID int64, model, // Fallback: if no exact prefix_hash match, aggregate by session_key from // routing_observations. This handles multi-turn sessions where the prefix // hash changes every turn but the session (and its cache behavior) is stable. - stats, fallbackErr := s.getSessionCacheStats(apiKeyHash, upstreamID, model, prefixHash, window, now) + stats, fallbackErr := s.getSessionCacheStats(apiKeyHash, upstreamID, model, prefixHash, protocol, window, now) if fallbackErr != nil { return stats, err } @@ -531,7 +560,7 @@ func (s *Store) GetPrefixCacheStats(apiKeyHash string, upstreamID int64, model, // getSessionCacheStats aggregates cache observations by session_key when the // exact prefix_hash doesn't exist in the summary table. This is the common case // for multi-turn conversations where each request has a slightly different prefix. -func (s *Store) getSessionCacheStats(apiKeyHash string, upstreamID int64, model, sessionKey string, window time.Duration, now time.Time) (PrefixCacheStats, error) { +func (s *Store) getSessionCacheStats(apiKeyHash string, upstreamID int64, model, sessionKey, protocol string, window time.Duration, now time.Time) (PrefixCacheStats, error) { stats := PrefixCacheStats{APIKeyHash: apiKeyHash, UpstreamID: upstreamID, Model: model, PrefixHash: sessionKey, SessionKey: sessionKey} from := now.Add(-window) err := s.queryRow(`SELECT @@ -592,14 +621,11 @@ func (s *Store) getSessionCacheStats(apiKeyHash string, upstreamID int64, model, if stats.LastCreatedAt > latestCache { latestCache = stats.LastCreatedAt } - // Use adaptive TTL for ExpiresAt: if the session has been running - // long enough and cache was rebuilt multiple times, assume 1h TTL - // was used for the latest creation (see selectCacheTTL in scheduler). - assumedTTL := 5 * time.Minute - sessionDuration := now.Unix() - stats.FirstSeenAt - if sessionDuration > int64((10*time.Minute)/time.Second) && stats.CreateCount >= 2 { - assumedTTL = time.Hour - } + // Use the same adaptive-TTL policy as routing.selectAdaptiveTTL so a + // Gemini session or a sparse conversation with any rebuild is treated + // as 1h, not 5min. The old 5-min-default flipped CacheHot → CacheExpired + // up to 55 minutes early, killing the guaranteed initial hit credit. + assumedTTL := AssumedCacheTTL(protocol, stats.Observations, stats.CreateCount, stats.FirstSeenAt, now.Unix()) stats.ExpiresAt = latestCache + int64(assumedTTL/time.Second) stats.Valid = stats.ExpiresAt > now.Unix() } diff --git a/internal/store/routing_persistence_test.go b/internal/store/routing_persistence_test.go index a18b759..3520b14 100644 --- a/internal/store/routing_persistence_test.go +++ b/internal/store/routing_persistence_test.go @@ -120,14 +120,14 @@ func TestRoutingObservationStatsAndCacheIsolation(t *testing.T) { if stats.P95TTFTMs != 200 || stats.P95DurationMs != 2000 { t.Fatalf("unexpected percentiles: %+v", stats) } - cache, err := st.GetPrefixCacheStats("key-a", 7, "m", "p", 15*time.Minute, now) + cache, err := st.GetPrefixCacheStats("key-a", 7, "m", "p", "claude", 15*time.Minute, now) if err != nil { t.Fatal(err) } if cache.HitCount != 1 || cache.MissCount != 1 || cache.CreateCount != 1 || cache.HitRate != 0.5 || !cache.Valid { t.Fatalf("unexpected key-a cache stats: %+v", cache) } - if _, err := st.GetPrefixCacheStats("key-b", 7, "m", "p", 15*time.Minute, now); err != nil { + if _, err := st.GetPrefixCacheStats("key-b", 7, "m", "p", "claude", 15*time.Minute, now); err != nil { // key-b has its own entry; reading it must not return key-a's cache. t.Fatal(err) } @@ -141,3 +141,33 @@ func TestRoutingObservationStatsAndCacheIsolation(t *testing.T) { t.Fatal(err) } } + + +// AssumedCacheTTL must mirror routing.selectAdaptiveTTL exactly, otherwise the +// store fallback would stamp an ExpiresAt derived from a wrong TTL and the +// state machine would flip CacheHot → CacheExpired prematurely. +func TestAssumedCacheTTL(t *testing.T) { + now := int64(1_700_000_000) + // Gemini always → 1h. + for _, p := range []string{"gemini", "google", "generativelanguage", "generatecontent"} { + if got := AssumedCacheTTL(p, 1, 1, now-60, now); got != time.Hour { + t.Fatalf("%s → %v want 1h", p, got) + } + } + // Long session + rebuilds → 1h. + if got := AssumedCacheTTL("claude", 20, 3, now-15*60, now); got != time.Hour { + t.Fatalf("long/rebuilds → %v want 1h", got) + } + // Sparse conversation + any rebuild → 1h. + if got := AssumedCacheTTL("claude", 2, 1, now-10*60, now); got != time.Hour { + t.Fatalf("sparse → %v want 1h", got) + } + // Short session, few rebuilds → 5min. + if got := AssumedCacheTTL("claude", 3, 1, now-3*60, now); got != 5*time.Minute { + t.Fatalf("short → %v want 5min", got) + } + // No observations → 5min default. + if got := AssumedCacheTTL("claude", 0, 0, 0, now); got != 5*time.Minute { + t.Fatalf("empty → %v want 5min", got) + } +} From 858260378250f546db58df89b8bd6e48495785ff Mon Sep 17 00:00:00 2001 From: Steven Moder Date: Fri, 11 Sep 2026 10:11:13 +0800 Subject: [PATCH 3/5] fix(store): symmetric success filter + surface fallback error in prefix cache stats MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related audit findings in the same fallback path: #4: getSessionCacheStats' in-window query counted all observations while its lifetime widen fallback filtered success=TRUE. A single failed retry inside the 15-min window suppressed the widen path — the same underlying history produced different verdicts depending on where the failure landed relative to the window boundary. Fix: apply success=TRUE to both queries (the widen path is now a strict superset of the window path), matching the comment's intent ('healthy history should survive circuit-breaker recoveries'). #5: GetPrefixCacheStats's fallback used ':=' inside the error block, shadowing 'stats' and 'fallbackErr'. On fallback failure it returned the primary err (typically ErrNoRows) and the shadowed partial stats, swallowing any real DB error from the fallback. Fix: propagate fallbackErr and return the outer initialised stats on the error path so callers/logs see the true failure and consistent field content. --- internal/store/routing_observations.go | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/internal/store/routing_observations.go b/internal/store/routing_observations.go index bac22cb..104b466 100644 --- a/internal/store/routing_observations.go +++ b/internal/store/routing_observations.go @@ -535,11 +535,16 @@ func (s *Store) GetPrefixCacheStats(apiKeyHash string, upstreamID int64, model, // Fallback: if no exact prefix_hash match, aggregate by session_key from // routing_observations. This handles multi-turn sessions where the prefix // hash changes every turn but the session (and its cache behavior) is stable. - stats, fallbackErr := s.getSessionCacheStats(apiKeyHash, upstreamID, model, prefixHash, protocol, window, now) + // + // When the fallback also fails, surface the fallback's error (which may + // carry a real DB failure) rather than the primary ErrNoRows that + // triggered the fallback. Return the outer initialised stats value so + // the caller sees consistent field content on the error path. + fallbackStats, fallbackErr := s.getSessionCacheStats(apiKeyHash, upstreamID, model, prefixHash, protocol, window, now) if fallbackErr != nil { - return stats, err + return stats, fallbackErr } - return stats, nil + return fallbackStats, nil } stats.HitRate = routingRatio(stats.HitCount, stats.HitCount+stats.MissCount) stats.Valid = stats.ExpiresAt > now.Unix() @@ -563,6 +568,11 @@ func (s *Store) GetPrefixCacheStats(apiKeyHash string, upstreamID int64, model, func (s *Store) getSessionCacheStats(apiKeyHash string, upstreamID int64, model, sessionKey, protocol string, window time.Duration, now time.Time) (PrefixCacheStats, error) { stats := PrefixCacheStats{APIKeyHash: apiKeyHash, UpstreamID: upstreamID, Model: model, PrefixHash: sessionKey, SessionKey: sessionKey} from := now.Add(-window) + // Both the in-window query and the lifetime fallback below MUST apply the + // same success=TRUE filter, otherwise a single failed retry inside the + // window suppresses the widen path and yields divergent verdicts for the + // same underlying history depending purely on where the failure lands + // relative to the window boundary. err := s.queryRow(`SELECT COUNT(*), COALESCE(SUM(CASE WHEN cache_hit THEN 1 ELSE 0 END),0), @@ -574,6 +584,7 @@ func (s *Store) getSessionCacheStats(apiKeyHash string, upstreamID int64, model, COALESCE(MIN(`+s.unixExpr("observed_at")+`),0) FROM routing_observations WHERE api_key_hash=? AND upstream_id=? AND model=? AND session_key=? + AND success=TRUE AND observed_at>=? AND observed_at Date: Fri, 11 Sep 2026 10:13:22 +0800 Subject: [PATCH 4/5] fix(routing): record actual_upstream_id so failover attempts are correctly attributed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit route_decisions previously stored only SelectedUpstreamID — the initial pick. When that pick failed and a failover attempt succeeded, the row's selected_upstream_id kept pointing at the failed upstream while actual_cost / actual_input_tokens / actual_cached_tokens described the successful failover attempt's usage. Any join on selected_upstream_id misattributed the request to the upstream that DID NOT serve it. Fix: add actual_upstream_id column (populated from result.FinalUpstreamID) alongside the existing actual_* usage columns. selected_upstream_id retains its meaning (initial pick) so historical decision reasoning stays intact. Downstream analytics that care about 'which upstream served this' should COALESCE(NULLIF(actual_upstream_id, 0), selected_upstream_id). Migration 20260911_030000 adds the column with default 0 (unknown). Fixes audit finding #8. --- ...30000_add_route_decisions_actual_upstream.sql | 5 +++++ internal/server/routing_audit.go | 9 ++++++++- internal/store/models_routing.go | 1 + internal/store/route_decisions.go | 16 +++++++++++++--- 4 files changed, 27 insertions(+), 4 deletions(-) create mode 100644 database/migrations/20260911_030000_add_route_decisions_actual_upstream.sql diff --git a/database/migrations/20260911_030000_add_route_decisions_actual_upstream.sql b/database/migrations/20260911_030000_add_route_decisions_actual_upstream.sql new file mode 100644 index 0000000..edc459e --- /dev/null +++ b/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; diff --git a/internal/server/routing_audit.go b/internal/server/routing_audit.go index 580dfd9..5ab2056 100644 --- a/internal/server/routing_audit.go +++ b/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) diff --git a/internal/store/models_routing.go b/internal/store/models_routing.go index 880dfee..9a0c99c 100644 --- a/internal/store/models_routing.go +++ b/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"` diff --git a/internal/store/route_decisions.go b/internal/store/route_decisions.go index 00258f0..b4e75cc 100644 --- a/internal/store/route_decisions.go +++ b/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 From d3de0cea801b34ed701faba79770511c340608b9 Mon Sep 17 00:00:00 2001 From: Steven Moder Date: Fri, 11 Sep 2026 10:18:28 +0800 Subject: [PATCH 5/5] fix(forward): normalize OpenAI/Gemini prompt_tokens to Anthropic uncached semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Anthropic responses report usage.input_tokens as EXCLUSIVE of cache_read / cache_creation. OpenAI (prompt_tokens) and Gemini (promptTokenCount) report their input as INCLUSIVE of cached. Previously parseUsageObject stored whichever was largest — mixing conventions in the same column. For any downstream consumer that sums the three fields to reconstruct the billed prompt total (e.g. usage attribution, cache-coverage analytics), this double-counts cached for OpenAI/Gemini rows by roughly cached/total. Fix at the ingest layer: parseUsageObject subtracts cached from the inclusive prompt_tokens/promptTokenCount, matching Anthropic semantics. Every column downstream (routing_observations.input_tokens, request_attempts.input_tokens) now has a single well-defined meaning. Tests: - TestUsageAuditNormalizesOpenAIPromptTokensToUncached - TestUsageAuditNormalizesGeminiPromptTokensToUncached - TestUsageAuditClampsNegativeUncached - Updated TestRelayResponseCapturesUsageBytesAndRequestID for new contract. --- internal/forward/audit.go | 34 ++++++++++++++++++++++++++++++++-- internal/forward/audit_test.go | 33 ++++++++++++++++++++++++++++++++- 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/internal/forward/audit.go b/internal/forward/audit.go index f27b1b0..6fdbfdc 100644 --- a/internal/forward/audit.go +++ b/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} } diff --git a/internal/forward/audit_test.go b/internal/forward/audit_test.go index 319a016..43635df 100644 --- a/internal/forward/audit_test.go +++ b/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" {