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/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" { 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) + } + } +} 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/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 diff --git a/internal/store/routing_observations.go b/internal/store/routing_observations.go index 849482a..104b466 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,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, 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() @@ -531,9 +565,14 @@ 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) + // 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), @@ -545,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 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) + } +}