diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 4c18ec8692..7336ee263b 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -8433,3 +8433,10 @@ update commands and credential authority unchanged. `internal/api/onboarding_handlers.go` changed only the human-readable pairing readiness messages so they name Settings > Pulse Mobile. No agent registration, lifecycle or install path changed. + +### AI usage export columns only + +`internal/api/ai_handlers.go` and `internal/api/chat_service_adapter.go` +changed only to carry prompt-cache token buckets through the AI usage export +and the Patrol stream response. No agent registration, enrolment, install, +update or removal path changed. diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 70d798f6ac..cb391dace3 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -8494,3 +8494,27 @@ turning AI back on needs no new path. This is presentation gating only: Patrol runtime, findings, schedules and the capability's server derivation are unchanged. `AppLayout.test.tsx`, `CommandPaletteModal.test.tsx`, `KeyboardShortcutsModal.test.tsx` and `App.architecture.test.ts` pin it. + +### Prompt-cache usage is accounted, not just logged + +Anthropic reports `cache_creation_input_tokens` and `cache_read_input_tokens` +as buckets disjoint from `input_tokens`. The provider client now carries both +buckets on `providers.ChatResponse` and the streaming `DoneEvent` (assigned +from `message_start` and the final cumulative `message_delta`, never summed +across events), the agentic loops and Patrol responses sum them beside the +ordinary totals, and the usage ledger (`cost.UsageEvent`), Patrol run records +and their persisted copies keep them. `cost.EstimateUsageUSD` prices each +bucket at the provider's cache rate when the table knows it (Anthropic: +5-minute write 1.25x and read 0.1x of the input rate) and at the ordinary +input rate otherwise; `cost.EstimateUSD` remains the no-cache wrapper, so +callers without cache data are unchanged. Spend summaries, the Patrol digest +spend, the cost projection (per-run medians now carry cache buckets) and the +usage export expose the buckets and price them the same way. `InputTokens` +keeps its meaning everywhere, so the telemetry token buckets and the +`patrol_input_tokens` semantics are unchanged; cache tokens are reported but +not folded into those counters. Proofs: `anthropic_test.go` (both response +paths keep the buckets, the final delta does not double them), +`pricing_anthropic_test.go`, `store_test.go`, `patrol_digest_test.go`, +`patrol_cost_projection_test.go` and `cost_persistence_test.go`. This repairs +the lost accounting data from Pulse #2350; it is not a measured provider bill +or a claim of savings across six-hourly runs. diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index deffbfb875..70e3a74fcc 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -11077,6 +11077,19 @@ boundaries and failures; `infrastructureAgentDoctorModel.test.ts` exercises the real lifecycle closures. These are modelled installer/privilege probes, not native system installation or publication. +### AI usage export carries prompt-cache buckets + +`GET /api/ai/cost/export` now includes `cache_creation_input_tokens` and +`cache_read_input_tokens` beside the existing input and output counts, in +the JSON events (as optional fields of the usage event) and as two CSV +columns after `output_tokens`, and prices each event with the cache-aware +estimator so `estimated_usd` matches the summary surfaces. Existing columns, +their order, the query parameters and the authentication boundary are +unchanged; the chat service adapter passes the same buckets through the +Patrol stream response without changing its request shape. +`ai_handlers_more_test.go` covers the export in both formats with a cached +Anthropic event. + ### Pulse Mobile pairing readiness wording The onboarding readiness diagnostics for a disabled or unconnected relay name diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index b261249403..d2ac577097 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -6342,3 +6342,12 @@ metrics target, health threshold, persistence or recovery authority changes. cover repeated paths/groups, direct children, scoped legacy snapshots, missing ownership and live target replacement. Synthetic presentation proof does not establish native appliance, installation or recovery acceptance. + +### Prompt-cache usage buckets open no storage or recovery path + +The AI usage ledger and Patrol run records gain two optional integer fields +(`cache_creation_input_tokens`, `cache_read_input_tokens`) in the files +Pulse already writes, with `omitempty` encoding so existing records load +unchanged and older builds ignore the fields. `internal/api/ai_handlers.go` +reads them for the usage export only. No new file, directory, retention +window, backup, migration or recovery authority is introduced. diff --git a/internal/ai/chat/agentic.go b/internal/ai/chat/agentic.go index 79921ad7f9..bdfa617b14 100644 --- a/internal/ai/chat/agentic.go +++ b/internal/ai/chat/agentic.go @@ -667,11 +667,13 @@ type AgenticLoop struct { modelName string // Token accumulation across all turns - totalInputTokens int - totalOutputTokens int - totalToolCalls int - totalModelTurns int - totalEvidenceCalls int + totalInputTokens int + totalOutputTokens int + totalCacheCreationTokens int + totalCacheReadTokens int + totalToolCalls int + totalModelTurns int + totalEvidenceCalls int // State for ongoing executions mu sync.Mutex @@ -1222,6 +1224,8 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me stopReason = data.StopReason toolCalls = agentcapabilities.NormalizeProviderToolCallsForExecution(data.ToolCalls) a.totalInputTokens += data.InputTokens + a.totalCacheCreationTokens += data.CacheCreationInputTokens + a.totalCacheReadTokens += data.CacheReadInputTokens a.totalOutputTokens += data.OutputTokens log.Info(). Int("turn", turn). diff --git a/internal/ai/chat/agentic_control.go b/internal/ai/chat/agentic_control.go index 12545d18c8..830c7c347c 100644 --- a/internal/ai/chat/agentic_control.go +++ b/internal/ai/chat/agentic_control.go @@ -165,6 +165,18 @@ func (a *AgenticLoop) GetTotalOutputTokens() int { return a.totalOutputTokens } +// GetTotalCacheCreationTokens returns the accumulated prompt-cache writes +// across all turns; disjoint from GetTotalInputTokens. +func (a *AgenticLoop) GetTotalCacheCreationTokens() int { + return a.totalCacheCreationTokens +} + +// GetTotalCacheReadTokens returns the accumulated prompt-cache reads across +// all turns; disjoint from GetTotalInputTokens. +func (a *AgenticLoop) GetTotalCacheReadTokens() int { + return a.totalCacheReadTokens +} + // GetTotalToolCalls returns the accepted model-selected tool call count across all turns. func (a *AgenticLoop) GetTotalToolCalls() int { return a.totalToolCalls @@ -186,5 +198,7 @@ func (a *AgenticLoop) GetTotalEvidenceCalls() int { func (a *AgenticLoop) ResetTokenCounts() { a.totalInputTokens = 0 a.totalOutputTokens = 0 + a.totalCacheCreationTokens = 0 + a.totalCacheReadTokens = 0 a.totalToolCalls = 0 } diff --git a/internal/ai/chat/agentic_final.go b/internal/ai/chat/agentic_final.go index a04f5a3411..8e1119b753 100644 --- a/internal/ai/chat/agentic_final.go +++ b/internal/ai/chat/agentic_final.go @@ -117,6 +117,10 @@ func (a *AgenticLoop) ensureFinalTextResponseWithSystemPrompt( case "done": if data, ok := event.Data.(providers.DoneEvent); ok { a.totalInputTokens += data.InputTokens + + a.totalCacheCreationTokens += data.CacheCreationInputTokens + + a.totalCacheReadTokens += data.CacheReadInputTokens a.totalOutputTokens += data.OutputTokens } } diff --git a/internal/ai/chat/service.go b/internal/ai/chat/service.go index 63f90189f1..b7a4edc7f1 100644 --- a/internal/ai/chat/service.go +++ b/internal/ai/chat/service.go @@ -371,7 +371,9 @@ func (s *Service) recordChatTurnCost(loop *AgenticLoop, requestModel, contextSco } inputTokens := loop.GetTotalInputTokens() outputTokens := loop.GetTotalOutputTokens() - if inputTokens == 0 && outputTokens == 0 { + cacheCreationTokens := loop.GetTotalCacheCreationTokens() + cacheReadTokens := loop.GetTotalCacheReadTokens() + if inputTokens == 0 && outputTokens == 0 && cacheCreationTokens == 0 && cacheReadTokens == 0 { return } providerName := "" @@ -381,15 +383,17 @@ func (s *Service) recordChatTurnCost(loop *AgenticLoop, requestModel, contextSco } } store.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: providerName, - RequestModel: requestModel, - UseCase: "chat", - ContextScope: strings.TrimSpace(contextScope), - ToolCallCount: loop.GetTotalToolCalls(), - InputTokens: inputTokens, - OutputTokens: outputTokens, - SessionID: strings.TrimSpace(sessionID), + Timestamp: time.Now(), + Provider: providerName, + RequestModel: requestModel, + UseCase: "chat", + ContextScope: strings.TrimSpace(contextScope), + ToolCallCount: loop.GetTotalToolCalls(), + InputTokens: inputTokens, + OutputTokens: outputTokens, + CacheCreationInputTokens: cacheCreationTokens, + CacheReadInputTokens: cacheReadTokens, + SessionID: strings.TrimSpace(sessionID), }) } @@ -2779,6 +2783,10 @@ type PatrolResponse struct { Content string `json:"content"` InputTokens int `json:"input_tokens"` OutputTokens int `json:"output_tokens"` + // Prompt-cache buckets summed across the loop's turns; disjoint from + // InputTokens, see providers.ChatResponse. + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` } // ExecutePatrolStream creates a temporary agentic loop for patrol execution. @@ -2924,8 +2932,10 @@ func (s *Service) ExecutePatrolStream(ctx context.Context, req PatrolRequest, ca } } return &PatrolResponse{ - InputTokens: tempLoop.GetTotalInputTokens(), - OutputTokens: tempLoop.GetTotalOutputTokens(), + InputTokens: tempLoop.GetTotalInputTokens(), + OutputTokens: tempLoop.GetTotalOutputTokens(), + CacheCreationInputTokens: tempLoop.GetTotalCacheCreationTokens(), + CacheReadInputTokens: tempLoop.GetTotalCacheReadTokens(), }, err } @@ -2971,9 +2981,11 @@ func (s *Service) ExecutePatrolStream(ctx context.Context, req PatrolRequest, ca callback(StreamEvent{Type: "done", Data: doneData}) return &PatrolResponse{ - Content: contentBuilder.String(), - InputTokens: tempLoop.GetTotalInputTokens(), - OutputTokens: tempLoop.GetTotalOutputTokens(), + Content: contentBuilder.String(), + InputTokens: tempLoop.GetTotalInputTokens(), + OutputTokens: tempLoop.GetTotalOutputTokens(), + CacheCreationInputTokens: tempLoop.GetTotalCacheCreationTokens(), + CacheReadInputTokens: tempLoop.GetTotalCacheReadTokens(), }, nil } diff --git a/internal/ai/chat/session_title.go b/internal/ai/chat/session_title.go index 7c04bcfe13..4f88ba5763 100644 --- a/internal/ai/chat/session_title.go +++ b/internal/ai/chat/session_title.go @@ -129,16 +129,18 @@ func (s *Service) generateSessionTitle(ctx context.Context, sessionID string) er } if costStore != nil && response != nil && (response.InputTokens != 0 || response.OutputTokens != 0) { costStore.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: provider.Name(), - RequestModel: requestModel, - ResponseModel: strings.TrimSpace(response.Model), - UseCase: "chat", - InputTokens: response.InputTokens, - OutputTokens: response.OutputTokens, - TargetType: "assistant_session_title", - TargetID: normalizedSessionID, - SessionID: normalizedSessionID, + Timestamp: time.Now(), + Provider: provider.Name(), + RequestModel: requestModel, + ResponseModel: strings.TrimSpace(response.Model), + UseCase: "chat", + InputTokens: response.InputTokens, + OutputTokens: response.OutputTokens, + CacheCreationInputTokens: response.CacheCreationInputTokens, + CacheReadInputTokens: response.CacheReadInputTokens, + TargetType: "assistant_session_title", + TargetID: normalizedSessionID, + SessionID: normalizedSessionID, }) } diff --git a/internal/ai/cost/pricing.go b/internal/ai/cost/pricing.go index c2717f18ca..488fe26e9d 100644 --- a/internal/ai/cost/pricing.go +++ b/internal/ai/cost/pricing.go @@ -7,19 +7,68 @@ import "strings" type TokenPrice struct { InputUSDPerMTok float64 OutputUSDPerMTok float64 - AsOf string + // CacheWriteUSDPerMTok and CacheReadUSDPerMTok price the prompt-cache + // buckets a provider reports beside ordinary input tokens. Zero means this + // table has no separate cache price for the provider, and the bucket is + // charged at the ordinary input rate, the conservative choice. + CacheWriteUSDPerMTok float64 + CacheReadUSDPerMTok float64 + AsOf string } +// TokenUsage is one call's token counts by billing bucket. InputTokens is the +// provider's ordinary, uncached input count; the cache buckets are disjoint +// from it, so PromptTokens is their sum. +type TokenUsage struct { + InputTokens int64 + OutputTokens int64 + CacheCreationInputTokens int64 + CacheReadInputTokens int64 +} + +// PromptTokens is every token sent to the model, cached or not. +func (u TokenUsage) PromptTokens() int64 { + return u.InputTokens + u.CacheCreationInputTokens + u.CacheReadInputTokens +} + +// Anthropic prices prompt-cache writes and reads as multiples of the model's +// input rate: a 5-minute cache write costs 1.25x and a cache read 0.1x +// (https://platform.claude.com/docs/en/about-claude/pricing, read 2026-10-03). +// Pulse only requests the 5-minute cache, so the 1-hour write rate is never +// charged. +const ( + anthropicCacheWriteMultiplier = 1.25 + anthropicCacheReadMultiplier = 0.10 +) + // EstimateUSD returns an estimated USD cost for the given provider/model and token counts. // If the model pricing is unknown, ok is false and usd is 0. func EstimateUSD(provider, model string, inputTokens, outputTokens int64) (usd float64, ok bool, price TokenPrice) { - price, ok = lookupPrice(provider, model, inputTokens) + return EstimateUsageUSD(provider, model, TokenUsage{InputTokens: inputTokens, OutputTokens: outputTokens}) +} + +// EstimateUsageUSD prices every bucket of a call's usage: ordinary input and +// output at the model's rates, and prompt-cache writes and reads at the +// provider's cache rates when the table knows them, otherwise at the input +// rate. Tier selection uses the whole prompt, cached or not, because that is +// the context size providers tier on. +func EstimateUsageUSD(provider, model string, usage TokenUsage) (usd float64, ok bool, price TokenPrice) { + price, ok = lookupPrice(provider, model, usage.PromptTokens()) if !ok { return 0, false, TokenPrice{} } - - usd = (float64(inputTokens)/1_000_000.0)*price.InputUSDPerMTok + - (float64(outputTokens)/1_000_000.0)*price.OutputUSDPerMTok + cacheWrite := price.CacheWriteUSDPerMTok + if cacheWrite <= 0 { + cacheWrite = price.InputUSDPerMTok + } + cacheRead := price.CacheReadUSDPerMTok + if cacheRead <= 0 { + cacheRead = price.InputUSDPerMTok + } + usd = (float64(usage.InputTokens)/1_000_000.0)*price.InputUSDPerMTok + + (float64(usage.OutputTokens)/1_000_000.0)*price.OutputUSDPerMTok + + (float64(usage.CacheCreationInputTokens)/1_000_000.0)*cacheWrite + + (float64(usage.CacheReadInputTokens)/1_000_000.0)*cacheRead return usd, true, price } @@ -173,11 +222,16 @@ func lookupPrice(provider, model string, inputTokens int64) (TokenPrice, bool) { if asOf == "" { asOf = pricingAsOf } - return TokenPrice{ + price := TokenPrice{ InputUSDPerMTok: tier.InputUSDPerMTok, OutputUSDPerMTok: tier.OutputUSDPerMTok, AsOf: asOf, - }, true + } + if provider == "anthropic" { + price.CacheWriteUSDPerMTok = tier.InputUSDPerMTok * anthropicCacheWriteMultiplier + price.CacheReadUSDPerMTok = tier.InputUSDPerMTok * anthropicCacheReadMultiplier + } + return price, true } } return TokenPrice{}, false diff --git a/internal/ai/cost/pricing_anthropic_test.go b/internal/ai/cost/pricing_anthropic_test.go index 89bd9a48be..0dd95ef6b0 100644 --- a/internal/ai/cost/pricing_anthropic_test.go +++ b/internal/ai/cost/pricing_anthropic_test.go @@ -1,6 +1,9 @@ package cost -import "testing" +import ( + "math" + "testing" +) func TestLookupPriceUsesCurrentAnthropicOpusPricing(t *testing.T) { tests := []struct { @@ -78,3 +81,46 @@ func TestLookupPriceUsesCurrentAnthropicGeneration5Pricing(t *testing.T) { }) } } + +func TestEstimateUsageUSDPricesAnthropicCacheBucketsAtCacheRates(t *testing.T) { + usage := TokenUsage{InputTokens: 100_000, OutputTokens: 10_000, CacheCreationInputTokens: 200_000, CacheReadInputTokens: 1_000_000} + usd, ok, price := EstimateUsageUSD("anthropic", "claude-sonnet-5", usage) + if !ok { + t.Fatal("expected known pricing for claude-sonnet-5") + } + // Sonnet 5 input is 2.00/M, so a 5-minute cache write is 2.50/M and a + // cache read 0.20/M. + if price.CacheWriteUSDPerMTok != 2.5 || price.CacheReadUSDPerMTok != 0.2 { + t.Fatalf("unexpected cache rates: %+v", price) + } + want := 0.1*2.00 + 0.01*10.00 + 0.2*2.50 + 1.0*0.20 + if math.Abs(usd-want) > 1e-9 { + t.Fatalf("usd = %f, want %f", usd, want) + } + // Charging the whole prompt at the input rate overstates a cached run. + flat, _, _ := EstimateUSD("anthropic", "claude-sonnet-5", usage.PromptTokens(), usage.OutputTokens) + if flat <= usd { + t.Fatalf("flat input-rate estimate %f should exceed cache-aware %f", flat, usd) + } +} + +func TestEstimateUsageUSDFallsBackToInputRateWithoutCachePrices(t *testing.T) { + usage := TokenUsage{InputTokens: 100_000, CacheReadInputTokens: 100_000} + usd, ok, price := EstimateUsageUSD("openai", "gpt-4o-mini", usage) + if !ok { + t.Fatal("expected known pricing for gpt-4o-mini") + } + if price.CacheWriteUSDPerMTok != 0 || price.CacheReadUSDPerMTok != 0 { + t.Fatalf("openai rows carry no cache prices: %+v", price) + } + // Both buckets at 0.15/M: conservative, never below the provider's bill. + if want := 0.2 * 0.15; math.Abs(usd-want) > 1e-9 { + t.Fatalf("usd = %f, want %f", usd, want) + } + // Without cache buckets the two estimators agree exactly. + a, _, _ := EstimateUSD("openai", "gpt-4o-mini", 100_000, 2_000) + b, _, _ := EstimateUsageUSD("openai", "gpt-4o-mini", TokenUsage{InputTokens: 100_000, OutputTokens: 2_000}) + if a != b { + t.Fatalf("EstimateUSD %f != EstimateUsageUSD %f", a, b) + } +} diff --git a/internal/ai/cost/store.go b/internal/ai/cost/store.go index b529c4bb94..4a50bf97a5 100644 --- a/internal/ai/cost/store.go +++ b/internal/ai/cost/store.go @@ -21,14 +21,28 @@ type UsageEvent struct { ToolCallCount int `json:"tool_call_count,omitempty"` InputTokens int `json:"input_tokens,omitempty"` OutputTokens int `json:"output_tokens,omitempty"` - TargetType string `json:"target_type,omitempty"` - TargetID string `json:"target_id,omitempty"` - FindingID string `json:"finding_id,omitempty"` + // CacheCreationInputTokens and CacheReadInputTokens are prompt-cache + // buckets reported beside InputTokens (Anthropic); they never overlap it. + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` + TargetType string `json:"target_type,omitempty"` + TargetID string `json:"target_id,omitempty"` + FindingID string `json:"finding_id,omitempty"` // SessionID ties chat-session spend (turns, compaction, title calls) // together so per-session cost can be summed for the drawer footer. SessionID string `json:"session_id,omitempty"` } +// TokenUsage returns the event's counts by billing bucket. +func (e UsageEvent) TokenUsage() TokenUsage { + return TokenUsage{ + InputTokens: int64(e.InputTokens), + OutputTokens: int64(e.OutputTokens), + CacheCreationInputTokens: int64(e.CacheCreationInputTokens), + CacheReadInputTokens: int64(e.CacheReadInputTokens), + } +} + // Persistence defines the storage contract for usage history. type Persistence interface { SaveUsageHistory(events []UsageEvent) error @@ -151,7 +165,7 @@ func (s *Store) SessionCostUSD(sessionID string) (usd float64, known bool) { } matched = true provider, model := ResolveProviderAndModel(e.Provider, e.RequestModel, e.ResponseModel) - eventUSD, eventKnown, _ := EstimateUSD(provider, model, int64(e.InputTokens), int64(e.OutputTokens)) + eventUSD, eventKnown, _ := EstimateUsageUSD(provider, model, e.TokenUsage()) if !eventKnown { return 0, false } @@ -231,7 +245,7 @@ func (s *Store) GetSummary(days int) Summary { pmTotals := make(map[pmKey]*ProviderModelSummary) dailyTotals := make(map[string]*DailySummary) - var totalInput, totalOutput int64 + var totalInput, totalOutput, totalCacheCreation, totalCacheRead int64 for _, e := range events { provider, model := ResolveProviderAndModel(e.Provider, e.RequestModel, e.ResponseModel) @@ -244,11 +258,15 @@ func (s *Store) GetSummary(days int) Summary { } pm.InputTokens += int64(e.InputTokens) pm.OutputTokens += int64(e.OutputTokens) + pm.CacheCreationInputTokens += int64(e.CacheCreationInputTokens) + pm.CacheReadInputTokens += int64(e.CacheReadInputTokens) totalInput += int64(e.InputTokens) totalOutput += int64(e.OutputTokens) + totalCacheCreation += int64(e.CacheCreationInputTokens) + totalCacheRead += int64(e.CacheReadInputTokens) - usd, known, _ := EstimateUSD(provider, model, int64(e.InputTokens), int64(e.OutputTokens)) + usd, known, _ := EstimateUsageUSD(provider, model, e.TokenUsage()) if known { pm.EstimatedUSD += usd pm.PricingKnown = true @@ -262,6 +280,8 @@ func (s *Store) GetSummary(days int) Summary { } ds.InputTokens += int64(e.InputTokens) ds.OutputTokens += int64(e.OutputTokens) + ds.CacheCreationInputTokens += int64(e.CacheCreationInputTokens) + ds.CacheReadInputTokens += int64(e.CacheReadInputTokens) if known { ds.EstimatedUSD += usd } @@ -269,7 +289,7 @@ func (s *Store) GetSummary(days int) Summary { providerModels := make([]ProviderModelSummary, 0, len(pmTotals)) for _, pm := range pmTotals { - pm.TotalTokens = pm.InputTokens + pm.OutputTokens + pm.TotalTokens = pm.InputTokens + pm.CacheCreationInputTokens + pm.CacheReadInputTokens + pm.OutputTokens providerModels = append(providerModels, *pm) } sort.Slice(providerModels, func(i, j int) bool { @@ -281,7 +301,7 @@ func (s *Store) GetSummary(days int) Summary { daily := make([]DailySummary, 0, len(dailyTotals)) for _, ds := range dailyTotals { - ds.TotalTokens = ds.InputTokens + ds.OutputTokens + ds.TotalTokens = ds.InputTokens + ds.CacheCreationInputTokens + ds.CacheReadInputTokens + ds.OutputTokens daily = append(daily, *ds) } sort.Slice(daily, func(i, j int) bool { @@ -289,10 +309,12 @@ func (s *Store) GetSummary(days int) Summary { }) totals := ProviderModelSummary{ - Provider: "all", - InputTokens: totalInput, - OutputTokens: totalOutput, - TotalTokens: totalInput + totalOutput, + Provider: "all", + InputTokens: totalInput, + OutputTokens: totalOutput, + CacheCreationInputTokens: totalCacheCreation, + CacheReadInputTokens: totalCacheRead, + TotalTokens: totalInput + totalCacheCreation + totalCacheRead + totalOutput, } for _, pm := range providerModels { @@ -415,44 +437,60 @@ func (s *Store) GetPersistenceStatus() (lastError error, lastSaveTime time.Time, // ProviderModelSummary is a rollup for a provider/model pair. type ProviderModelSummary struct { - Provider string `json:"provider"` - Model string `json:"model"` - InputTokens int64 `json:"input_tokens"` - OutputTokens int64 `json:"output_tokens"` - TotalTokens int64 `json:"total_tokens"` - EstimatedUSD float64 `json:"estimated_usd,omitempty"` - PricingKnown bool `json:"pricing_known"` + Provider string `json:"provider"` + Model string `json:"model"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + // Prompt-cache buckets, disjoint from InputTokens; TotalTokens counts + // every token in and out, cached or not. + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int64 `json:"cache_read_input_tokens,omitempty"` + TotalTokens int64 `json:"total_tokens"` + EstimatedUSD float64 `json:"estimated_usd,omitempty"` + PricingKnown bool `json:"pricing_known"` } // DailySummary is a rollup for a single day across all providers. type DailySummary struct { - Date string `json:"date"` - InputTokens int64 `json:"input_tokens"` - OutputTokens int64 `json:"output_tokens"` - TotalTokens int64 `json:"total_tokens"` - EstimatedUSD float64 `json:"estimated_usd,omitempty"` + Date string `json:"date"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + // Prompt-cache buckets, disjoint from InputTokens; TotalTokens counts + // every token in and out, cached or not. + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int64 `json:"cache_read_input_tokens,omitempty"` + TotalTokens int64 `json:"total_tokens"` + EstimatedUSD float64 `json:"estimated_usd,omitempty"` } // UseCaseSummary is a rollup for a use-case (e.g. "chat", "patrol"). type UseCaseSummary struct { - UseCase string `json:"use_case"` - InputTokens int64 `json:"input_tokens"` - OutputTokens int64 `json:"output_tokens"` - TotalTokens int64 `json:"total_tokens"` - EstimatedUSD float64 `json:"estimated_usd,omitempty"` - PricingKnown bool `json:"pricing_known"` + UseCase string `json:"use_case"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + // Prompt-cache buckets, disjoint from InputTokens; TotalTokens counts + // every token in and out, cached or not. + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int64 `json:"cache_read_input_tokens,omitempty"` + TotalTokens int64 `json:"total_tokens"` + EstimatedUSD float64 `json:"estimated_usd,omitempty"` + PricingKnown bool `json:"pricing_known"` } // TargetSummary is a rollup for a Pulse target (e.g. vm/container/node). type TargetSummary struct { - TargetType string `json:"target_type"` - TargetID string `json:"target_id"` - Calls int64 `json:"calls"` - InputTokens int64 `json:"input_tokens"` - OutputTokens int64 `json:"output_tokens"` - TotalTokens int64 `json:"total_tokens"` - EstimatedUSD float64 `json:"estimated_usd,omitempty"` - PricingKnown bool `json:"pricing_known"` + TargetType string `json:"target_type"` + TargetID string `json:"target_id"` + Calls int64 `json:"calls"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + // Prompt-cache buckets, disjoint from InputTokens; TotalTokens counts + // every token in and out, cached or not. + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int64 `json:"cache_read_input_tokens,omitempty"` + TotalTokens int64 `json:"total_tokens"` + EstimatedUSD float64 `json:"estimated_usd,omitempty"` + PricingKnown bool `json:"pricing_known"` } // Summary is returned by the cost summary API. @@ -473,10 +511,12 @@ type Summary struct { func summarizeUseCases(events []UsageEvent) []UseCaseSummary { type totals struct { - input int64 - output int64 - usd float64 - known bool + input int64 + output int64 + cacheCreation int64 + cacheRead int64 + usd float64 + known bool } perUseCase := make(map[string]*totals) @@ -493,10 +533,12 @@ func summarizeUseCases(events []UsageEvent) []UseCaseSummary { t.input += int64(e.InputTokens) t.output += int64(e.OutputTokens) + t.cacheCreation += int64(e.CacheCreationInputTokens) + t.cacheRead += int64(e.CacheReadInputTokens) provider, model := ResolveProviderAndModel(e.Provider, e.RequestModel, e.ResponseModel) - usd, known, _ := EstimateUSD(provider, model, int64(e.InputTokens), int64(e.OutputTokens)) + usd, known, _ := EstimateUsageUSD(provider, model, e.TokenUsage()) if known { t.usd += usd t.known = true @@ -506,12 +548,14 @@ func summarizeUseCases(events []UsageEvent) []UseCaseSummary { out := make([]UseCaseSummary, 0, len(perUseCase)) for useCase, t := range perUseCase { out = append(out, UseCaseSummary{ - UseCase: useCase, - InputTokens: t.input, - OutputTokens: t.output, - TotalTokens: t.input + t.output, - EstimatedUSD: t.usd, - PricingKnown: t.known, + UseCase: useCase, + InputTokens: t.input, + OutputTokens: t.output, + CacheCreationInputTokens: t.cacheCreation, + CacheReadInputTokens: t.cacheRead, + TotalTokens: t.input + t.cacheCreation + t.cacheRead + t.output, + EstimatedUSD: t.usd, + PricingKnown: t.known, }) } @@ -540,11 +584,13 @@ func summarizeTargets(events []UsageEvent) []TargetSummary { targetID string } type totals struct { - calls int64 - input int64 - output int64 - usd float64 - known bool + calls int64 + input int64 + output int64 + cacheCreation int64 + cacheRead int64 + usd float64 + known bool } perTarget := make(map[key]*totals) @@ -563,9 +609,11 @@ func summarizeTargets(events []UsageEvent) []TargetSummary { t.calls++ t.input += int64(e.InputTokens) t.output += int64(e.OutputTokens) + t.cacheCreation += int64(e.CacheCreationInputTokens) + t.cacheRead += int64(e.CacheReadInputTokens) provider, model := ResolveProviderAndModel(e.Provider, e.RequestModel, e.ResponseModel) - usd, known, _ := EstimateUSD(provider, model, int64(e.InputTokens), int64(e.OutputTokens)) + usd, known, _ := EstimateUsageUSD(provider, model, e.TokenUsage()) if known { t.usd += usd t.known = true @@ -575,14 +623,16 @@ func summarizeTargets(events []UsageEvent) []TargetSummary { out := make([]TargetSummary, 0, len(perTarget)) for k, t := range perTarget { out = append(out, TargetSummary{ - TargetType: k.targetType, - TargetID: k.targetID, - Calls: t.calls, - InputTokens: t.input, - OutputTokens: t.output, - TotalTokens: t.input + t.output, - EstimatedUSD: t.usd, - PricingKnown: t.known, + TargetType: k.targetType, + TargetID: k.targetID, + Calls: t.calls, + InputTokens: t.input, + OutputTokens: t.output, + CacheCreationInputTokens: t.cacheCreation, + CacheReadInputTokens: t.cacheRead, + TotalTokens: t.input + t.cacheCreation + t.cacheRead + t.output, + EstimatedUSD: t.usd, + PricingKnown: t.known, }) } diff --git a/internal/ai/cost/store_test.go b/internal/ai/cost/store_test.go index ca2cb0934f..a16bc4d57b 100644 --- a/internal/ai/cost/store_test.go +++ b/internal/ai/cost/store_test.go @@ -540,3 +540,55 @@ func TestSessionCostUSD(t *testing.T) { t.Errorf("expected free session to be known at 0, got usd=%v known=%v", usd, known) } } + +func TestSummaryCountsAndPricesPromptCacheBuckets(t *testing.T) { + store := NewStore(90) + now := time.Now() + store.Record(UsageEvent{ + Timestamp: now, + Provider: "anthropic", + RequestModel: "anthropic:claude-sonnet-5", + UseCase: "patrol", + TargetType: "node", + TargetID: "pve1", + InputTokens: 100_000, + OutputTokens: 10_000, + CacheCreationInputTokens: 200_000, + CacheReadInputTokens: 1_000_000, + }) + + // 0.1M*2.00 + 0.01M*10.00 + 0.2M*2.50 + 1.0M*0.20 + wantUSD := 0.2 + 0.1 + 0.5 + 0.2 + check := func(name string, in, out, creation, read, total int64, usd float64, known bool) { + t.Helper() + if in != 100_000 || out != 10_000 || creation != 200_000 || read != 1_000_000 { + t.Fatalf("%s buckets = in %d out %d creation %d read %d", name, in, out, creation, read) + } + if total != 1_310_000 { + t.Fatalf("%s total tokens = %d, want every bucket counted", name, total) + } + if !known || math.Abs(usd-wantUSD) > 1e-9 { + t.Fatalf("%s estimated usd = %f (known=%v), want %f", name, usd, known, wantUSD) + } + } + + summary := store.GetSummary(7) + tot := summary.Totals + check("totals", tot.InputTokens, tot.OutputTokens, tot.CacheCreationInputTokens, tot.CacheReadInputTokens, tot.TotalTokens, tot.EstimatedUSD, tot.PricingKnown) + if len(summary.ProviderModels) != 1 || len(summary.UseCases) != 1 || len(summary.Targets) != 1 || len(summary.DailyTotals) != 1 { + t.Fatalf("unexpected summary shape: %+v", summary) + } + pm := summary.ProviderModels[0] + check("provider model", pm.InputTokens, pm.OutputTokens, pm.CacheCreationInputTokens, pm.CacheReadInputTokens, pm.TotalTokens, pm.EstimatedUSD, pm.PricingKnown) + uc := summary.UseCases[0] + check("use case", uc.InputTokens, uc.OutputTokens, uc.CacheCreationInputTokens, uc.CacheReadInputTokens, uc.TotalTokens, uc.EstimatedUSD, uc.PricingKnown) + tg := summary.Targets[0] + check("target", tg.InputTokens, tg.OutputTokens, tg.CacheCreationInputTokens, tg.CacheReadInputTokens, tg.TotalTokens, tg.EstimatedUSD, tg.PricingKnown) + ds := summary.DailyTotals[0] + check("daily", ds.InputTokens, ds.OutputTokens, ds.CacheCreationInputTokens, ds.CacheReadInputTokens, ds.TotalTokens, ds.EstimatedUSD, true) + + events := store.ListEvents(7) + if len(events) != 1 || events[0].CacheCreationInputTokens != 200_000 || events[0].CacheReadInputTokens != 1_000_000 { + t.Fatalf("listed event lost cache buckets: %+v", events) + } +} diff --git a/internal/ai/cost_persistence.go b/internal/ai/cost_persistence.go index d57edcce48..74b506a306 100644 --- a/internal/ai/cost_persistence.go +++ b/internal/ai/cost_persistence.go @@ -25,18 +25,20 @@ func (a *CostPersistenceAdapter) SaveUsageHistory(events []cost.UsageEvent) erro records := make([]config.AIUsageEventRecord, len(events)) for i, e := range events { records[i] = config.AIUsageEventRecord{ - Timestamp: e.Timestamp, - Provider: e.Provider, - RequestModel: e.RequestModel, - ResponseModel: e.ResponseModel, - UseCase: e.UseCase, - ContextScope: e.ContextScope, - ToolCallCount: e.ToolCallCount, - InputTokens: e.InputTokens, - OutputTokens: e.OutputTokens, - TargetType: e.TargetType, - TargetID: e.TargetID, - FindingID: e.FindingID, + Timestamp: e.Timestamp, + Provider: e.Provider, + RequestModel: e.RequestModel, + ResponseModel: e.ResponseModel, + UseCase: e.UseCase, + ContextScope: e.ContextScope, + ToolCallCount: e.ToolCallCount, + InputTokens: e.InputTokens, + OutputTokens: e.OutputTokens, + CacheCreationInputTokens: e.CacheCreationInputTokens, + CacheReadInputTokens: e.CacheReadInputTokens, + TargetType: e.TargetType, + TargetID: e.TargetID, + FindingID: e.FindingID, } } return a.config.SaveAIUsageHistory(records) @@ -55,18 +57,20 @@ func (a *CostPersistenceAdapter) LoadUsageHistory() ([]cost.UsageEvent, error) { events := make([]cost.UsageEvent, len(data.Events)) for i, r := range data.Events { events[i] = cost.UsageEvent{ - Timestamp: r.Timestamp, - Provider: r.Provider, - RequestModel: r.RequestModel, - ResponseModel: r.ResponseModel, - UseCase: r.UseCase, - ContextScope: r.ContextScope, - ToolCallCount: r.ToolCallCount, - InputTokens: r.InputTokens, - OutputTokens: r.OutputTokens, - TargetType: r.TargetType, - TargetID: r.TargetID, - FindingID: r.FindingID, + Timestamp: r.Timestamp, + Provider: r.Provider, + RequestModel: r.RequestModel, + ResponseModel: r.ResponseModel, + UseCase: r.UseCase, + ContextScope: r.ContextScope, + ToolCallCount: r.ToolCallCount, + InputTokens: r.InputTokens, + OutputTokens: r.OutputTokens, + CacheCreationInputTokens: r.CacheCreationInputTokens, + CacheReadInputTokens: r.CacheReadInputTokens, + TargetType: r.TargetType, + TargetID: r.TargetID, + FindingID: r.FindingID, } } return events, nil diff --git a/internal/ai/cost_persistence_test.go b/internal/ai/cost_persistence_test.go index 3065dcd9af..607c0102f3 100644 --- a/internal/ai/cost_persistence_test.go +++ b/internal/ai/cost_persistence_test.go @@ -149,3 +149,27 @@ func TestCostPersistenceAdapter_NilConfigFailsClosed(t *testing.T) { t.Fatal("expected load to return normalized empty slice on failure") } } + +func TestCostPersistenceAdapterRoundTripsPromptCacheBuckets(t *testing.T) { + adapter := NewCostPersistenceAdapter(config.NewConfigPersistence(t.TempDir())) + saved := []cost.UsageEvent{{ + Timestamp: time.Now(), + Provider: "anthropic", + RequestModel: "anthropic:claude-sonnet-5", + UseCase: "patrol", + InputTokens: 100, + OutputTokens: 20, + CacheCreationInputTokens: 1000, + CacheReadInputTokens: 5000, + }} + if err := adapter.SaveUsageHistory(saved); err != nil { + t.Fatalf("SaveUsageHistory: %v", err) + } + loaded, err := adapter.LoadUsageHistory() + if err != nil { + t.Fatalf("LoadUsageHistory: %v", err) + } + if len(loaded) != 1 || loaded[0].CacheCreationInputTokens != 1000 || loaded[0].CacheReadInputTokens != 5000 || loaded[0].InputTokens != 100 { + t.Fatalf("round trip lost cache buckets: %+v", loaded) + } +} diff --git a/internal/ai/patrol.go b/internal/ai/patrol.go index 068d9441b0..eb93e20b73 100644 --- a/internal/ai/patrol.go +++ b/internal/ai/patrol.go @@ -162,9 +162,11 @@ type PatrolRunRecord struct { TriageFlags int `json:"triage_flags"` // Number of deterministic flags found TriageSkippedLLM bool `json:"triage_skipped_llm,omitempty"` // Legacy: true for older quiet-triage runs // AI Analysis details - AIAnalysis string `json:"ai_analysis,omitempty"` // Display-safe AI run analysis summary - InputTokens int `json:"input_tokens,omitempty"` // Tokens sent to AI - OutputTokens int `json:"output_tokens,omitempty"` // Tokens received from AI + AIAnalysis string `json:"ai_analysis,omitempty"` // Display-safe AI run analysis summary + InputTokens int `json:"input_tokens,omitempty"` // Tokens sent to AI + OutputTokens int `json:"output_tokens,omitempty"` // Tokens received from AI + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` // Prompt-cache writes, disjoint from InputTokens + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` // Prompt-cache reads, disjoint from InputTokens // Tool call traces ToolCalls []ToolCallRecord `json:"tool_calls,omitempty"` ToolCallCount int `json:"tool_call_count"` @@ -211,6 +213,8 @@ type patrolRunRecordJSON struct { AIAnalysis string `json:"ai_analysis,omitempty"` InputTokens int `json:"input_tokens,omitempty"` OutputTokens int `json:"output_tokens,omitempty"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` // Prompt-cache writes, disjoint from InputTokens + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` // Prompt-cache reads, disjoint from InputTokens ToolCalls []ToolCallRecord `json:"tool_calls,omitempty"` ToolCallCount int `json:"tool_call_count"` } @@ -343,6 +347,8 @@ func (r PatrolRunRecord) MarshalJSON() ([]byte, error) { AIAnalysis: normalized.AIAnalysis, InputTokens: normalized.InputTokens, OutputTokens: normalized.OutputTokens, + CacheCreationInputTokens: normalized.CacheCreationInputTokens, + CacheReadInputTokens: normalized.CacheReadInputTokens, ToolCalls: normalized.ToolCalls, ToolCallCount: normalized.ToolCallCount, }) @@ -396,6 +402,8 @@ func (r *PatrolRunRecord) UnmarshalJSON(data []byte) error { AIAnalysis: payload.AIAnalysis, InputTokens: payload.InputTokens, OutputTokens: payload.OutputTokens, + CacheCreationInputTokens: payload.CacheCreationInputTokens, + CacheReadInputTokens: payload.CacheReadInputTokens, ToolCalls: payload.ToolCalls, ToolCallCount: payload.ToolCallCount, }) diff --git a/internal/ai/patrol_ai.go b/internal/ai/patrol_ai.go index 02e1bf151d..d770cc8641 100644 --- a/internal/ai/patrol_ai.go +++ b/internal/ai/patrol_ai.go @@ -34,20 +34,23 @@ import ( // AIAnalysisResult contains the results of an AI analysis type AIAnalysisResult struct { - Response string // The AI's raw response text - Findings []*Finding // Parsed findings from the response - RejectedFindings int // Findings rejected by threshold validation - TriageFlags int // Number of deterministic triage flags - TriageSkippedLLM bool // Legacy: true for older records where quiet triage skipped LLM - InputTokens int - OutputTokens int - ToolCalls []ToolCallRecord // Tool invocations during this analysis - ReportedIDs []string // Finding IDs reported (created/re-reported) this run - NewFindingIDs []string // Finding IDs first created during this run - ResolvedIDs []string // Finding IDs explicitly resolved by LLM this run - Assessments []PatrolFindingAssessment // Explicit verdicts for existing findings this run - SeededFindingIDs []string // Finding IDs that were presented in seed context - QueriedFindingIDs []string // Finding IDs returned by patrol_get_findings this run + Response string // The AI's raw response text + Findings []*Finding // Parsed findings from the response + RejectedFindings int // Findings rejected by threshold validation + TriageFlags int // Number of deterministic triage flags + TriageSkippedLLM bool // Legacy: true for older records where quiet triage skipped LLM + InputTokens int + OutputTokens int + // Prompt-cache buckets, disjoint from InputTokens (providers.ChatResponse). + CacheCreationInputTokens int + CacheReadInputTokens int + ToolCalls []ToolCallRecord // Tool invocations during this analysis + ReportedIDs []string // Finding IDs reported (created/re-reported) this run + NewFindingIDs []string // Finding IDs first created during this run + ResolvedIDs []string // Finding IDs explicitly resolved by LLM this run + Assessments []PatrolFindingAssessment // Explicit verdicts for existing findings this run + SeededFindingIDs []string // Finding IDs that were presented in seed context + QueriedFindingIDs []string // Finding IDs returned by patrol_get_findings this run // Forecasts are the deterministic capacity forecasts computed this run, // stamped onto matching findings so the surface shows a first-class // urgency signal instead of relying on model prose. Carried as the @@ -802,26 +805,28 @@ func (p *PatrolService) runAIAnalysisState(ctx context.Context, snap patrolRunti } return attempt, chatErr } - buildAnalysisResult := func(content string, toolCalls []ToolCallRecord, inputTokens, outputTokens int) *AIAnalysisResult { + buildAnalysisResult := func(content string, toolCalls []ToolCallRecord, inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens int) *AIAnalysisResult { adapter.findingsMu.Lock() rejectedCount := adapter.rejectedCount adapter.findingsMu.Unlock() return &AIAnalysisResult{ - Response: CleanThinkingTokens(content), - Findings: adapter.getCollectedFindings(), - RejectedFindings: rejectedCount, - TriageFlags: len(triageResult.Flags), - TriageSkippedLLM: false, - InputTokens: inputTokens, - OutputTokens: outputTokens, - ToolCalls: append([]ToolCallRecord(nil), toolCalls...), - ReportedIDs: adapter.getReportedFindingIDs(), - NewFindingIDs: adapter.getNewFindingIDs(), - ResolvedIDs: adapter.getResolvedIDs(), - Assessments: adapter.getAssessments(), - SeededFindingIDs: seededFindingIDs, - QueriedFindingIDs: adapter.getQueriedFindingIDs(), - Forecasts: triageResult.Intel.forecasts, + Response: CleanThinkingTokens(content), + Findings: adapter.getCollectedFindings(), + RejectedFindings: rejectedCount, + TriageFlags: len(triageResult.Flags), + TriageSkippedLLM: false, + InputTokens: inputTokens, + OutputTokens: outputTokens, + CacheCreationInputTokens: cacheCreationTokens, + CacheReadInputTokens: cacheReadTokens, + ToolCalls: append([]ToolCallRecord(nil), toolCalls...), + ReportedIDs: adapter.getReportedFindingIDs(), + NewFindingIDs: adapter.getNewFindingIDs(), + ResolvedIDs: adapter.getResolvedIDs(), + Assessments: adapter.getAssessments(), + SeededFindingIDs: seededFindingIDs, + QueriedFindingIDs: adapter.getQueriedFindingIDs(), + Forecasts: triageResult.Intel.forecasts, } } @@ -859,15 +864,17 @@ func (p *PatrolService) runAIAnalysisState(ctx context.Context, snap patrolRunti if chatErr != nil { var partialResult *AIAnalysisResult if attempt != nil && attempt.response != nil { - p.recordPatrolUsage(attempt.response.InputTokens, attempt.response.OutputTokens) + p.recordPatrolUsage(attempt.response.InputTokens, attempt.response.OutputTokens, attempt.response.CacheCreationInputTokens, attempt.response.CacheReadInputTokens) } if attempt != nil { - inputTokens, outputTokens := 0, 0 + inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens := 0, 0, 0, 0 if attempt.response != nil { inputTokens = attempt.response.InputTokens outputTokens = attempt.response.OutputTokens + cacheCreationTokens = attempt.response.CacheCreationInputTokens + cacheReadTokens = attempt.response.CacheReadInputTokens } - candidate := buildAnalysisResult(attempt.finalContent, attempt.toolCalls, inputTokens, outputTokens) + candidate := buildAnalysisResult(attempt.finalContent, attempt.toolCalls, inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens) if candidate.Response != "" || inputTokens > 0 || outputTokens > 0 || len(candidate.ToolCalls) > 0 || len(candidate.Findings) > 0 || len(candidate.ResolvedIDs) > 0 || len(candidate.Assessments) > 0 || len(candidate.QueriedFindingIDs) > 0 { partialResult = candidate @@ -883,7 +890,9 @@ func (p *PatrolService) runAIAnalysisState(ctx context.Context, snap patrolRunti finalContent := attempt.finalContent inputTokens = attempt.response.InputTokens outputTokens = attempt.response.OutputTokens - p.recordPatrolUsage(attempt.response.InputTokens, attempt.response.OutputTokens) + cacheCreationTokens := attempt.response.CacheCreationInputTokens + cacheReadTokens := attempt.response.CacheReadInputTokens + p.recordPatrolUsage(attempt.response.InputTokens, attempt.response.OutputTokens, attempt.response.CacheCreationInputTokens, attempt.response.CacheReadInputTokens) log.Debug(). Int("input_tokens", inputTokens). @@ -900,7 +909,7 @@ func (p *PatrolService) runAIAnalysisState(ctx context.Context, snap patrolRunti // The original conversation owns the diagnosis and finding decisions. // Omitted assessments remain explicit in the result. A separate model // session must not infer them from old finding excerpts or signal counts. - return buildAnalysisResult(finalContent, attempt.toolCalls, inputTokens, outputTokens), nil + return buildAnalysisResult(finalContent, attempt.toolCalls, inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens), nil } // These are execution limits, independent of heuristic flags or inventory size. @@ -912,7 +921,7 @@ func patrolDetectionMaxTurns(scope *PatrolScope) int { return 40 } -func (p *PatrolService) recordPatrolUsage(inputTokens, outputTokens int) { +func (p *PatrolService) recordPatrolUsage(inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens int) { if p == nil || p.aiService == nil || (inputTokens <= 0 && outputTokens <= 0) { return } @@ -947,12 +956,14 @@ func (p *PatrolService) recordPatrolUsage(inputTokens, outputTokens int) { } store.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: providerName, - RequestModel: model, - UseCase: "patrol", - InputTokens: inputTokens, - OutputTokens: outputTokens, + Timestamp: time.Now(), + Provider: providerName, + RequestModel: model, + UseCase: "patrol", + InputTokens: inputTokens, + OutputTokens: outputTokens, + CacheCreationInputTokens: cacheCreationTokens, + CacheReadInputTokens: cacheReadTokens, }) } diff --git a/internal/ai/patrol_cost_projection.go b/internal/ai/patrol_cost_projection.go index 1154456bab..93c4ad00a3 100644 --- a/internal/ai/patrol_cost_projection.go +++ b/internal/ai/patrol_cost_projection.go @@ -120,12 +120,21 @@ type PatrolCostProjection struct { PricingAsOf string `json:"pricing_as_of,omitempty"` InputUSDPerMTok float64 `json:"input_usd_per_mtok"` OutputUSDPerMTok float64 `json:"output_usd_per_mtok"` + // Cache rates are reported when the provider prices prompt-cache writes + // and reads separately (Anthropic); zero means the input rate applies. + CacheWriteUSDPerMTok float64 `json:"cache_write_usd_per_mtok,omitempty"` + CacheReadUSDPerMTok float64 `json:"cache_read_usd_per_mtok,omitempty"` - PerRunInputTokens int64 `json:"per_run_input_tokens"` - PerRunOutputTokens int64 `json:"per_run_output_tokens"` - PerRunSource string `json:"per_run_source"` - HistoryRunCount int `json:"history_run_count"` - PerRunUSD float64 `json:"per_run_usd"` + PerRunInputTokens int64 `json:"per_run_input_tokens"` + PerRunOutputTokens int64 `json:"per_run_output_tokens"` + // Per-run prompt-cache buckets from the install's own history, disjoint + // from PerRunInputTokens. The measured default has none because it was + // taken from an uncached run. + PerRunCacheCreationInputTokens int64 `json:"per_run_cache_creation_input_tokens,omitempty"` + PerRunCacheReadInputTokens int64 `json:"per_run_cache_read_input_tokens,omitempty"` + PerRunSource string `json:"per_run_source"` + HistoryRunCount int `json:"history_run_count"` + PerRunUSD float64 `json:"per_run_usd"` IntervalMinutes int `json:"interval_minutes"` ScheduledRunsPerDay float64 `json:"scheduled_runs_per_day"` @@ -221,7 +230,8 @@ func ProjectPatrolCost(in PatrolCostProjectionInput) PatrolCostProjection { // enough priced history, else the measured default. windowStart := now.AddDate(0, 0, -patrolCostProjectionWindowDays) oldest := now - var fullInputs, fullOutputs, trigInputs, trigOutputs []int64 + var fullInputs, fullOutputs, fullCacheWrites, fullCacheReads []int64 + var trigInputs, trigOutputs, trigCacheWrites, trigCacheReads []int64 triggeredCount := 0 for _, run := range in.Runs { if run.InputTokens <= 0 || run.StartedAt.Before(windowStart) { @@ -233,11 +243,15 @@ func ProjectPatrolCost(in PatrolCostProjectionInput) PatrolCostProjection { if patrolRunIsFull(run) { fullInputs = append(fullInputs, int64(run.InputTokens)) fullOutputs = append(fullOutputs, int64(run.OutputTokens)) + fullCacheWrites = append(fullCacheWrites, int64(run.CacheCreationInputTokens)) + fullCacheReads = append(fullCacheReads, int64(run.CacheReadInputTokens)) continue } triggeredCount++ trigInputs = append(trigInputs, int64(run.InputTokens)) trigOutputs = append(trigOutputs, int64(run.OutputTokens)) + trigCacheWrites = append(trigCacheWrites, int64(run.CacheCreationInputTokens)) + trigCacheReads = append(trigCacheReads, int64(run.CacheReadInputTokens)) } out.PerRunSource = PatrolCostPerRunSourceDefault out.PerRunInputTokens = DefaultPatrolRunInputTokens @@ -246,6 +260,8 @@ func ProjectPatrolCost(in PatrolCostProjectionInput) PatrolCostProjection { out.PerRunSource = PatrolCostPerRunSourceHistory out.PerRunInputTokens = medianInt64(fullInputs) out.PerRunOutputTokens = medianInt64(fullOutputs) + out.PerRunCacheCreationInputTokens = medianInt64(fullCacheWrites) + out.PerRunCacheReadInputTokens = medianInt64(fullCacheReads) out.HistoryRunCount = len(fullInputs) } @@ -267,24 +283,37 @@ func ProjectPatrolCost(in PatrolCostProjectionInput) PatrolCostProjection { } // Pricing. - _, known, price := cost.EstimateUSD(provider, model, out.PerRunInputTokens, out.PerRunOutputTokens) + perRunUsage := cost.TokenUsage{ + InputTokens: out.PerRunInputTokens, + OutputTokens: out.PerRunOutputTokens, + CacheCreationInputTokens: out.PerRunCacheCreationInputTokens, + CacheReadInputTokens: out.PerRunCacheReadInputTokens, + } + _, known, price := cost.EstimateUsageUSD(provider, model, perRunUsage) out.PricingKnown = known if known { out.PricingAsOf = price.AsOf out.InputUSDPerMTok = price.InputUSDPerMTok out.OutputUSDPerMTok = price.OutputUSDPerMTok + out.CacheWriteUSDPerMTok = price.CacheWriteUSDPerMTok + out.CacheReadUSDPerMTok = price.CacheReadUSDPerMTok out.BilledPerToken = price.InputUSDPerMTok > 0 || price.OutputUSDPerMTok > 0 } - perRun := func(inputTokens, outputTokens int64) float64 { + perRun := func(usage cost.TokenUsage) float64 { if !known { return 0 } - usd, _, _ := cost.EstimateUSD(provider, model, inputTokens, outputTokens) + usd, _, _ := cost.EstimateUsageUSD(provider, model, usage) return usd } - out.PerRunUSD = roundUSD(perRun(out.PerRunInputTokens, out.PerRunOutputTokens)) + out.PerRunUSD = roundUSD(perRun(perRunUsage)) if triggeredCount > 0 { - out.TriggeredPerRunUSD = roundUSD(perRun(medianInt64(trigInputs), medianInt64(trigOutputs))) + out.TriggeredPerRunUSD = roundUSD(perRun(cost.TokenUsage{ + InputTokens: medianInt64(trigInputs), + OutputTokens: medianInt64(trigOutputs), + CacheCreationInputTokens: medianInt64(trigCacheWrites), + CacheReadInputTokens: medianInt64(trigCacheReads), + })) } scheduled30d := func(intervalMinutes int) float64 { if intervalMinutes <= 0 { diff --git a/internal/ai/patrol_cost_projection_test.go b/internal/ai/patrol_cost_projection_test.go index c31098272c..f889fcd6b0 100644 --- a/internal/ai/patrol_cost_projection_test.go +++ b/internal/ai/patrol_cost_projection_test.go @@ -231,3 +231,33 @@ func TestBlockOnExhaustedBudget_SetsPatrolBlockedState(t *testing.T) { t.Fatalf("non-budget failures must not block, got %q", other.lastBlockedCause) } } + +func TestProjectPatrolCost_HistoryCarriesPromptCacheBuckets(t *testing.T) { + now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC) + runs := []PatrolRunRecord{ + {StartedAt: now.Add(-6 * time.Hour), TriggerReason: string(TriggerReasonScheduled), InputTokens: 10_000, OutputTokens: 4_000, CacheCreationInputTokens: 20_000, CacheReadInputTokens: 100_000}, + {StartedAt: now.Add(-12 * time.Hour), TriggerReason: string(TriggerReasonScheduled), InputTokens: 12_000, OutputTokens: 4_000, CacheCreationInputTokens: 22_000, CacheReadInputTokens: 110_000}, + {StartedAt: now.Add(-18 * time.Hour), TriggerReason: string(TriggerReasonScheduled), InputTokens: 14_000, OutputTokens: 4_000, CacheCreationInputTokens: 24_000, CacheReadInputTokens: 120_000}, + } + got := ProjectPatrolCost(PatrolCostProjectionInput{ + Provider: "anthropic", + Model: "claude-sonnet-5", + IntervalMinutes: 360, + BudgetUSD30d: 20, + Runs: runs, + Now: now, + }) + if got.PerRunSource != PatrolCostPerRunSourceHistory || got.HistoryRunCount != 3 { + t.Fatalf("expected history-based estimate, got %+v", got) + } + if got.PerRunInputTokens != 12_000 || got.PerRunCacheCreationInputTokens != 22_000 || got.PerRunCacheReadInputTokens != 110_000 { + t.Fatalf("expected median cache buckets, got in=%d creation=%d read=%d", got.PerRunInputTokens, got.PerRunCacheCreationInputTokens, got.PerRunCacheReadInputTokens) + } + if got.CacheWriteUSDPerMTok != 2.5 || got.CacheReadUSDPerMTok != 0.2 { + t.Fatalf("expected Anthropic cache rates on the projection, got %+v", got) + } + // 0.012M*2.00 + 0.004M*10.00 + 0.022M*2.50 + 0.11M*0.20 = 0.141 per run. + if got.PerRunUSD < 0.140 || got.PerRunUSD > 0.142 { + t.Fatalf("per-run cost %.4f should price cache reads at a tenth of input", got.PerRunUSD) + } +} diff --git a/internal/ai/patrol_digest.go b/internal/ai/patrol_digest.go index 330b385dde..1a97e19772 100644 --- a/internal/ai/patrol_digest.go +++ b/internal/ai/patrol_digest.go @@ -113,7 +113,11 @@ type PatrolDigestSpend struct { PricingKnown bool `json:"pricing_known"` InputTokens int64 `json:"input_tokens"` OutputTokens int64 `json:"output_tokens"` - Calls int `json:"calls"` + // Prompt-cache buckets, disjoint from InputTokens and priced at the + // provider's cache rates when known. + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int64 `json:"cache_read_input_tokens,omitempty"` + Calls int `json:"calls"` } // PatrolDigest is the wire payload for GET /api/ai/patrol/digest. @@ -403,8 +407,10 @@ func summarizeDigestSpend(events []cost.UsageEvent, inWindow func(time.Time) boo summary.Calls++ summary.InputTokens += int64(event.InputTokens) summary.OutputTokens += int64(event.OutputTokens) + summary.CacheCreationInputTokens += int64(event.CacheCreationInputTokens) + summary.CacheReadInputTokens += int64(event.CacheReadInputTokens) provider, model := cost.ResolveProviderAndModel(event.Provider, event.RequestModel, event.ResponseModel) - usd, known, _ := cost.EstimateUSD(provider, model, int64(event.InputTokens), int64(event.OutputTokens)) + usd, known, _ := cost.EstimateUsageUSD(provider, model, event.TokenUsage()) if !known { summary.PricingKnown = false continue diff --git a/internal/ai/patrol_digest_test.go b/internal/ai/patrol_digest_test.go index 051b0a74f6..aea8169251 100644 --- a/internal/ai/patrol_digest_test.go +++ b/internal/ai/patrol_digest_test.go @@ -1,6 +1,7 @@ package ai import ( + "math" "testing" "time" @@ -196,3 +197,25 @@ func TestBuildPatrolDigestEmptyInputIsZeroNotNil(t *testing.T) { t.Fatalf("outcome order = %v", order) } } + +func TestSummarizeDigestSpendPricesPromptCacheBuckets(t *testing.T) { + t.Parallel() + now := time.Now() + events := []cost.UsageEvent{ + {Timestamp: now, Provider: "anthropic", RequestModel: "anthropic:claude-sonnet-5", UseCase: "patrol", + InputTokens: 100_000, OutputTokens: 10_000, CacheCreationInputTokens: 200_000, CacheReadInputTokens: 1_000_000}, + // Not Patrol: ignored. + {Timestamp: now, Provider: "anthropic", RequestModel: "anthropic:claude-sonnet-5", UseCase: "chat", InputTokens: 5}, + } + spend := summarizeDigestSpend(events, func(time.Time) bool { return true }) + if spend.Calls != 1 || spend.InputTokens != 100_000 || spend.OutputTokens != 10_000 { + t.Fatalf("spend = %+v", spend) + } + if spend.CacheCreationInputTokens != 200_000 || spend.CacheReadInputTokens != 1_000_000 { + t.Fatalf("cache buckets missing from digest spend: %+v", spend) + } + // Priced at Anthropic cache rates, not the input rate for every token. + if want := 0.2 + 0.1 + 0.5 + 0.2; !spend.PricingKnown || math.Abs(spend.EstimatedUSD-want) > 1e-9 { + t.Fatalf("estimated usd = %f (known=%v), want %f", spend.EstimatedUSD, spend.PricingKnown, want) + } +} diff --git a/internal/ai/patrol_history_persistence.go b/internal/ai/patrol_history_persistence.go index 85b7bf0441..b298f2ae1a 100644 --- a/internal/ai/patrol_history_persistence.go +++ b/internal/ai/patrol_history_persistence.go @@ -78,6 +78,8 @@ func (a *PatrolHistoryPersistenceAdapter) SavePatrolRunHistory(runs []PatrolRunR AIAnalysis: normalized.AIAnalysis, InputTokens: normalized.InputTokens, OutputTokens: normalized.OutputTokens, + CacheCreationInputTokens: normalized.CacheCreationInputTokens, + CacheReadInputTokens: normalized.CacheReadInputTokens, ToolCalls: convertAIToolCallsToConfig(normalized.ToolCalls), ToolCallCount: normalized.ToolCallCount, }) @@ -137,6 +139,8 @@ func (a *PatrolHistoryPersistenceAdapter) LoadPatrolRunHistory() ([]PatrolRunRec AIAnalysis: r.AIAnalysis, InputTokens: r.InputTokens, OutputTokens: r.OutputTokens, + CacheCreationInputTokens: r.CacheCreationInputTokens, + CacheReadInputTokens: r.CacheReadInputTokens, ToolCalls: convertConfigToolCallsToAI(r.ToolCalls), ToolCallCount: r.ToolCallCount, }) diff --git a/internal/ai/patrol_model_readiness.go b/internal/ai/patrol_model_readiness.go index fe4ae53a09..91fd0c4d12 100644 --- a/internal/ai/patrol_model_readiness.go +++ b/internal/ai/patrol_model_readiness.go @@ -127,14 +127,16 @@ type PatrolModelReadinessResult struct { // Details carries per-scenario probe and validator evidence ("expected // exactly one tool call, got 0", "nonce did not match"). Without it a // failed evaluation is undiagnosable from the result alone (#1624, #1614). - Details []string `json:"details,omitempty"` - Metadata *PatrolModelReadinessMetadata `json:"metadata,omitempty"` - Dimensions PatrolModelReadinessDimensions `json:"dimensions"` - Modes PatrolModelReadinessModes `json:"modes"` - CacheKey string `json:"-"` - inputTokens int - outputTokens int - providerCalls int + Details []string `json:"details,omitempty"` + Metadata *PatrolModelReadinessMetadata `json:"metadata,omitempty"` + Dimensions PatrolModelReadinessDimensions `json:"dimensions"` + Modes PatrolModelReadinessModes `json:"modes"` + CacheKey string `json:"-"` + inputTokens int + outputTokens int + cacheCreationTokens int + cacheReadTokens int + providerCalls int } type patrolModelReadinessCache struct { @@ -583,13 +585,15 @@ type patrolReadinessScenario struct { } type patrolReadinessProbeOutcome struct { - toolCalls []providers.ToolCall - stopReason string - duration time.Duration - firstResponse time.Duration - inputTokens int - outputTokens int - completionEvent bool + toolCalls []providers.ToolCall + stopReason string + duration time.Duration + firstResponse time.Duration + inputTokens int + outputTokens int + cacheCreationTokens int + cacheReadTokens int + completionEvent bool } func runPatrolModelReadinessWithProvider(ctx context.Context, cfg *config.AIConfig, providerName, model, modelString string, provider providers.Provider) PatrolModelReadinessResult { @@ -706,6 +710,8 @@ func runPatrolModelReadinessWithProvider(ctx context.Context, cfg *config.AIConf outcome, err := runPatrolReadinessStreamProbe(ctx, streamingProvider, req) result.inputTokens += outcome.inputTokens result.outputTokens += outcome.outputTokens + result.cacheCreationTokens += outcome.cacheCreationTokens + result.cacheReadTokens += outcome.cacheReadTokens if err != nil { probeErr = err details = append(details, fmt.Sprintf("Scenario %q probe failed: %v", scenario.name, err)) @@ -780,6 +786,8 @@ func runPatrolModelReadinessWithProvider(ctx context.Context, cfg *config.AIConf outcome, err := runPatrolReadinessStreamProbe(ctx, streamingProvider, followup) result.inputTokens += outcome.inputTokens result.outputTokens += outcome.outputTokens + result.cacheCreationTokens += outcome.cacheCreationTokens + result.cacheReadTokens += outcome.cacheReadTokens if err != nil { probeErr = err details = append(details, fmt.Sprintf("Multi-turn continuation probe failed: %v", err)) @@ -1106,12 +1114,16 @@ func runPatrolReadinessStreamProbe(ctx context.Context, provider providers.Strea outcome.stopReason = data.StopReason outcome.inputTokens = data.InputTokens outcome.outputTokens = data.OutputTokens + outcome.cacheCreationTokens = data.CacheCreationInputTokens + outcome.cacheReadTokens = data.CacheReadInputTokens outcome.completionEvent = true } else if data, ok := event.Data.(*providers.DoneEvent); ok && data != nil { outcome.toolCalls = append([]providers.ToolCall(nil), data.ToolCalls...) outcome.stopReason = data.StopReason outcome.inputTokens = data.InputTokens outcome.outputTokens = data.OutputTokens + outcome.cacheCreationTokens = data.CacheCreationInputTokens + outcome.cacheReadTokens = data.CacheReadInputTokens outcome.completionEvent = true } } @@ -1261,11 +1273,13 @@ func (s *Service) recordPatrolModelReadinessUsage(result PatrolModelReadinessRes return } costStore.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: result.Provider, - RequestModel: modelStringForReadinessResult(result), - UseCase: "patrol_readiness", - InputTokens: result.inputTokens, - OutputTokens: result.outputTokens, + Timestamp: time.Now(), + Provider: result.Provider, + RequestModel: modelStringForReadinessResult(result), + UseCase: "patrol_readiness", + InputTokens: result.inputTokens, + OutputTokens: result.outputTokens, + CacheCreationInputTokens: result.cacheCreationTokens, + CacheReadInputTokens: result.cacheReadTokens, }) } diff --git a/internal/ai/patrol_run.go b/internal/ai/patrol_run.go index fb499b5c4f..67c08b5e60 100644 --- a/internal/ai/patrol_run.go +++ b/internal/ai/patrol_run.go @@ -797,6 +797,8 @@ func (p *PatrolService) runPatrolWithTriggerStart(ctx context.Context, trigger T runRecord.FindingAssessments = append([]PatrolFindingAssessment(nil), runStats.aiAnalysis.Assessments...) runRecord.InputTokens = runStats.aiAnalysis.InputTokens runRecord.OutputTokens = runStats.aiAnalysis.OutputTokens + runRecord.CacheCreationInputTokens = runStats.aiAnalysis.CacheCreationInputTokens + runRecord.CacheReadInputTokens = runStats.aiAnalysis.CacheReadInputTokens runRecord.TriageFlags = runStats.triageFlags runRecord.TriageSkippedLLM = runStats.triageSkippedLLM toolCalls := runStats.aiAnalysis.ToolCalls @@ -1266,6 +1268,8 @@ func (p *PatrolService) runScopedPatrolWithStart(ctx context.Context, scope Patr runRecord.FindingAssessments = append([]PatrolFindingAssessment(nil), runStats.aiAnalysis.Assessments...) runRecord.InputTokens = runStats.aiAnalysis.InputTokens runRecord.OutputTokens = runStats.aiAnalysis.OutputTokens + runRecord.CacheCreationInputTokens = runStats.aiAnalysis.CacheCreationInputTokens + runRecord.CacheReadInputTokens = runStats.aiAnalysis.CacheReadInputTokens runRecord.TriageFlags = runStats.triageFlags runRecord.TriageSkippedLLM = runStats.triageSkippedLLM toolCalls := runStats.aiAnalysis.ToolCalls diff --git a/internal/ai/providers/anthropic.go b/internal/ai/providers/anthropic.go index b4ba514002..3ccb08554c 100644 --- a/internal/ai/providers/anthropic.go +++ b/internal/ai/providers/anthropic.go @@ -483,6 +483,10 @@ func (c *AnthropicClient) Chat(ctx context.Context, req ChatRequest) (*ChatRespo ToolCalls: toolCalls, InputTokens: anthropicResp.Usage.InputTokens, OutputTokens: anthropicResp.Usage.OutputTokens, + // Cache buckets are billed at their own rates and are not part of + // input_tokens, so they travel separately into cost accounting. + CacheCreationInputTokens: anthropicResp.Usage.CacheCreationInputTokens, + CacheReadInputTokens: anthropicResp.Usage.CacheReadInputTokens, }, nil } @@ -622,6 +626,23 @@ type anthropicDelta struct { StopReason string `json:"stop_reason,omitempty"` } +// captureAnthropicPromptUsage copies the prompt-side usage counters from a +// stream event. Anthropic reports input_tokens, cache_creation_input_tokens +// and cache_read_input_tokens as disjoint buckets and repeats them as +// cumulative values in message_start and the final message_delta, so each is +// assigned when present, never summed across events. +func captureAnthropicPromptUsage(usage anthropicUsage, input, cacheCreation, cacheRead *int) { + if usage.InputTokens > 0 { + *input = usage.InputTokens + } + if usage.CacheCreationInputTokens > 0 { + *cacheCreation = usage.CacheCreationInputTokens + } + if usage.CacheReadInputTokens > 0 { + *cacheRead = usage.CacheReadInputTokens + } +} + // ChatStream sends a chat request and streams the response via callback func (c *AnthropicClient) ChatStream(ctx context.Context, req ChatRequest, callback StreamCallback) error { // Convert messages to Anthropic format (same as Chat) @@ -735,7 +756,7 @@ func (c *AnthropicClient) ChatStream(ctx context.Context, req ChatRequest, callb var currentToolID string var currentToolName string var currentToolInput strings.Builder - var inputTokens, outputTokens int + var inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens int for { n, err := reader.Read(buf) @@ -769,8 +790,8 @@ func (c *AnthropicClient) ChatStream(ctx context.Context, req ChatRequest, callb switch event.Type { case "message_start": - if event.Message != nil && event.Message.Usage.InputTokens > 0 { - inputTokens = event.Message.Usage.InputTokens + if event.Message != nil { + captureAnthropicPromptUsage(event.Message.Usage, &inputTokens, &cacheCreationTokens, &cacheReadTokens) } case "content_block_start": @@ -838,8 +859,11 @@ func (c *AnthropicClient) ChatStream(ctx context.Context, req ChatRequest, callb } case "message_delta": - if event.Delta != nil && event.Delta.StopReason != "" { - if event.Usage != nil { + if event.Usage != nil { + // The final message_delta repeats the prompt-side usage + // as cumulative values, so this assigns rather than adds. + captureAnthropicPromptUsage(*event.Usage, &inputTokens, &cacheCreationTokens, &cacheReadTokens) + if event.Delta != nil && event.Delta.StopReason != "" { outputTokens = event.Usage.OutputTokens } } @@ -852,10 +876,12 @@ func (c *AnthropicClient) ChatStream(ctx context.Context, req ChatRequest, callb callback(StreamEvent{ Type: "done", Data: DoneEvent{ - StopReason: stopReason, - ToolCalls: toolCalls, - InputTokens: inputTokens, - OutputTokens: outputTokens, + StopReason: stopReason, + ToolCalls: toolCalls, + InputTokens: inputTokens, + OutputTokens: outputTokens, + CacheCreationInputTokens: cacheCreationTokens, + CacheReadInputTokens: cacheReadTokens, }, }) diff --git a/internal/ai/providers/anthropic_test.go b/internal/ai/providers/anthropic_test.go index 4879e59931..c0e647b33d 100644 --- a/internal/ai/providers/anthropic_test.go +++ b/internal/ai/providers/anthropic_test.go @@ -902,3 +902,87 @@ func TestAnthropicClient_ChatStream_CacheableSystemPrefix(t *testing.T) { t.Fatalf("tool breakpoint must be dropped when the system block is cached: %+v", got.Tools[0]) } } + +// Issue #2350: cache-creation and cache-read usage must survive both response +// paths so cost accounting can price each bucket at its own rate. +func TestAnthropicClient_Chat_KeepsPromptCacheUsageBuckets(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(anthropicResponse{ + ID: "msg_cache", + Type: "message", + Role: "assistant", + Model: "claude-sonnet-5", + StopReason: "end_turn", + Content: []anthropicContent{{Type: "text", Text: "ok"}}, + Usage: anthropicUsage{ + InputTokens: 100, + CacheCreationInputTokens: 1000, + CacheReadInputTokens: 5000, + OutputTokens: 20, + }, + }) + })) + defer server.Close() + + client := NewAnthropicClientWithBaseURL("test-key", "claude-sonnet-5", server.URL, 0) + resp, err := client.Chat(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "Hi"}}, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + if resp.InputTokens != 100 || resp.OutputTokens != 20 { + t.Fatalf("ordinary usage changed: in=%d out=%d", resp.InputTokens, resp.OutputTokens) + } + if resp.CacheCreationInputTokens != 1000 || resp.CacheReadInputTokens != 5000 { + t.Fatalf("cache buckets dropped: creation=%d read=%d", resp.CacheCreationInputTokens, resp.CacheReadInputTokens) + } +} + +func TestAnthropicClient_ChatStream_KeepsPromptCacheUsageBuckets(t *testing.T) { + // message_start carries the prompt-side usage; the final message_delta + // repeats it as cumulative values, which must assign, not double. + stream := []string{ + `{"type":"message_start","message":{"usage":{"input_tokens":100,"cache_creation_input_tokens":1000,"cache_read_input_tokens":5000,"output_tokens":1}}}`, + `{"type":"content_block_start","content_block":{"type":"text"}}`, + `{"type":"content_block_delta","delta":{"type":"text_delta","text":"ok"}}`, + `{"type":"content_block_stop"}`, + `{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":100,"cache_creation_input_tokens":1000,"cache_read_input_tokens":5000,"output_tokens":20}}`, + `{"type":"message_stop"}`, + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + for _, event := range stream { + _, _ = w.Write([]byte("data: " + event + "\n\n")) + w.(http.Flusher).Flush() + } + })) + defer server.Close() + + client := NewAnthropicClientWithBaseURL("test-key", "claude-sonnet-5", server.URL, 0) + var done DoneEvent + doneCalled := false + err := client.ChatStream(context.Background(), ChatRequest{ + Messages: []Message{{Role: "user", Content: "Hi"}}, + }, func(event StreamEvent) { + if event.Type == "done" { + if data, ok := event.Data.(DoneEvent); ok { + done = data + doneCalled = true + } + } + }) + if err != nil { + t.Fatalf("ChatStream: %v", err) + } + if !doneCalled { + t.Fatal("no done event") + } + if done.InputTokens != 100 || done.OutputTokens != 20 { + t.Fatalf("ordinary usage changed: in=%d out=%d", done.InputTokens, done.OutputTokens) + } + if done.CacheCreationInputTokens != 1000 || done.CacheReadInputTokens != 5000 { + t.Fatalf("cache buckets dropped or doubled: creation=%d read=%d", done.CacheCreationInputTokens, done.CacheReadInputTokens) + } +} diff --git a/internal/ai/providers/provider.go b/internal/ai/providers/provider.go index 86c9d7e13c..5f28dade53 100644 --- a/internal/ai/providers/provider.go +++ b/internal/ai/providers/provider.go @@ -146,6 +146,12 @@ type ChatResponse struct { ToolCalls []ToolCall `json:"tool_calls"` // Tool invocations InputTokens int `json:"input_tokens,omitempty"` OutputTokens int `json:"output_tokens,omitempty"` + // CacheCreationInputTokens and CacheReadInputTokens are the prompt-cache + // buckets a provider reports beside InputTokens (Anthropic). InputTokens + // stays the provider's ordinary, uncached input count, so the three never + // overlap; the cost ledger prices each bucket at its own rate. + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` } func (r ChatResponse) NormalizeCollections() ChatResponse { @@ -273,6 +279,9 @@ type DoneEvent struct { ToolCalls []ToolCall `json:"tool_calls"` InputTokens int `json:"input_tokens,omitempty"` OutputTokens int `json:"output_tokens,omitempty"` + // Prompt-cache buckets, disjoint from InputTokens; see ChatResponse. + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` } func (e DoneEvent) NormalizeCollections() DoneEvent { diff --git a/internal/ai/report_fleet_narrator.go b/internal/ai/report_fleet_narrator.go index 354974e556..3e6343c6bb 100644 --- a/internal/ai/report_fleet_narrator.go +++ b/internal/ai/report_fleet_narrator.go @@ -204,15 +204,17 @@ func (s *Service) NarrateFleet(ctx context.Context, in reporting.FleetNarrativeI providerName = provider.Name() } costStore.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: providerName, - RequestModel: model, - ResponseModel: resp.Model, - UseCase: reportFleetNarratorUseCase, - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, - TargetType: "fleet", - TargetID: strings.TrimSpace(in.Title), + Timestamp: time.Now(), + Provider: providerName, + RequestModel: model, + ResponseModel: resp.Model, + UseCase: reportFleetNarratorUseCase, + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + CacheCreationInputTokens: resp.CacheCreationInputTokens, + CacheReadInputTokens: resp.CacheReadInputTokens, + TargetType: "fleet", + TargetID: strings.TrimSpace(in.Title), }) } diff --git a/internal/ai/report_narrator.go b/internal/ai/report_narrator.go index 0f083cee9b..8ea890cbc0 100644 --- a/internal/ai/report_narrator.go +++ b/internal/ai/report_narrator.go @@ -225,15 +225,17 @@ func (s *Service) Narrate(ctx context.Context, in reporting.NarrativeInput) (rep providerName = provider.Name() } costStore.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: providerName, - RequestModel: model, - ResponseModel: resp.Model, - UseCase: reportNarratorUseCase, - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, - TargetType: in.ResourceType, - TargetID: in.ResourceID, + Timestamp: time.Now(), + Provider: providerName, + RequestModel: model, + ResponseModel: resp.Model, + UseCase: reportNarratorUseCase, + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + CacheCreationInputTokens: resp.CacheCreationInputTokens, + CacheReadInputTokens: resp.CacheReadInputTokens, + TargetType: in.ResourceType, + TargetID: in.ResourceID, }) } diff --git a/internal/ai/service.go b/internal/ai/service.go index 13a637f48a..ddd974b733 100644 --- a/internal/ai/service.go +++ b/internal/ai/service.go @@ -226,6 +226,10 @@ type PatrolStreamResponse struct { Content string `json:"content"` InputTokens int `json:"input_tokens"` OutputTokens int `json:"output_tokens"` + // Prompt-cache buckets summed across the run's provider calls; disjoint + // from InputTokens, see providers.ChatResponse. + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` // ToolCalls carries calls observed by bounded Patrol follow-up passes so // the parent run's durable audit contains every provider action. ToolCalls []ToolCallRecord `json:"tool_calls,omitempty"` @@ -1978,14 +1982,16 @@ func (s *Service) QuickAnalysis(ctx context.Context, req QuickAnalysisRequest) ( providerName = provider.Name() } costStore.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: providerName, - RequestModel: model, - ResponseModel: resp.Model, - UseCase: useCase, - TargetType: strings.TrimSpace(req.TargetType), - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, + Timestamp: time.Now(), + Provider: providerName, + RequestModel: model, + ResponseModel: resp.Model, + UseCase: useCase, + TargetType: strings.TrimSpace(req.TargetType), + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + CacheCreationInputTokens: resp.CacheCreationInputTokens, + CacheReadInputTokens: resp.CacheReadInputTokens, }) } @@ -2169,12 +2175,16 @@ func assistantContextScopeForExecuteRequest(req ExecuteRequest) string { // ExecuteResponse represents the AI's response type ExecuteResponse struct { - Content string `json:"content"` - Model string `json:"model"` - InputTokens int `json:"input_tokens"` - OutputTokens int `json:"output_tokens"` - ToolCalls []ToolExecution `json:"tool_calls"` // Commands that were executed - PendingApprovals []ApprovalNeededData `json:"pending_approvals"` // Commands that require approval (non-streaming) + Content string `json:"content"` + Model string `json:"model"` + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + // Prompt-cache buckets summed across the loop's provider calls; disjoint + // from InputTokens, see providers.ChatResponse. + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` + ToolCalls []ToolExecution `json:"tool_calls"` // Commands that were executed + PendingApprovals []ApprovalNeededData `json:"pending_approvals"` // Commands that require approval (non-streaming) } func EmptyExecuteResponse() ExecuteResponse { @@ -2357,6 +2367,8 @@ Always execute the commands rather than telling the user how to do it.` var toolExecutions []ToolExecution totalInputTokens := 0 totalOutputTokens := 0 + totalCacheCreationTokens := 0 + totalCacheReadTokens := 0 var finalContent string var model string @@ -2393,22 +2405,26 @@ Always execute the commands rather than telling the user how to do it.` providerName = provider.Name() } costStore.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: providerName, - RequestModel: modelString, - ResponseModel: resp.Model, - UseCase: req.UseCase, - ContextScope: assistantContextScopeForExecuteRequest(req), - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, - TargetType: req.TargetType, - TargetID: req.TargetID, - FindingID: req.FindingID, + Timestamp: time.Now(), + Provider: providerName, + RequestModel: modelString, + ResponseModel: resp.Model, + UseCase: req.UseCase, + ContextScope: assistantContextScopeForExecuteRequest(req), + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + CacheCreationInputTokens: resp.CacheCreationInputTokens, + CacheReadInputTokens: resp.CacheReadInputTokens, + TargetType: req.TargetType, + TargetID: req.TargetID, + FindingID: req.FindingID, }) } totalInputTokens += resp.InputTokens totalOutputTokens += resp.OutputTokens + totalCacheCreationTokens += resp.CacheCreationInputTokens + totalCacheReadTokens += resp.CacheReadInputTokens model = resp.Model finalContent = resp.Content @@ -2452,22 +2468,26 @@ Always execute the commands rather than telling the user how to do it.` // The caller can execute approvals via /api/ai/run-command and continue. if len(pendingApprovals) > 0 { return &ExecuteResponse{ - Content: finalContent, - Model: model, - InputTokens: totalInputTokens, - OutputTokens: totalOutputTokens, - ToolCalls: toolExecutions, - PendingApprovals: pendingApprovals, + Content: finalContent, + Model: model, + InputTokens: totalInputTokens, + OutputTokens: totalOutputTokens, + CacheCreationInputTokens: totalCacheCreationTokens, + CacheReadInputTokens: totalCacheReadTokens, + ToolCalls: toolExecutions, + PendingApprovals: pendingApprovals, }, nil } } return &ExecuteResponse{ - Content: finalContent, - Model: model, - InputTokens: totalInputTokens, - OutputTokens: totalOutputTokens, - ToolCalls: toolExecutions, + Content: finalContent, + Model: model, + InputTokens: totalInputTokens, + OutputTokens: totalOutputTokens, + CacheCreationInputTokens: totalCacheCreationTokens, + CacheReadInputTokens: totalCacheReadTokens, + ToolCalls: toolExecutions, }, nil } @@ -2583,6 +2603,8 @@ Always execute the commands rather than telling the user how to do it.` var toolExecutions []ToolExecution totalInputTokens := 0 totalOutputTokens := 0 + totalCacheCreationTokens := 0 + totalCacheReadTokens := 0 var finalContent string var model string @@ -2636,17 +2658,19 @@ Always execute the commands rather than telling the user how to do it.` providerName = provider.Name() } costStore.Record(cost.UsageEvent{ - Timestamp: time.Now(), - Provider: providerName, - RequestModel: modelString, - ResponseModel: resp.Model, - UseCase: req.UseCase, - ContextScope: assistantContextScopeForExecuteRequest(req), - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, - TargetType: req.TargetType, - TargetID: req.TargetID, - FindingID: req.FindingID, + Timestamp: time.Now(), + Provider: providerName, + RequestModel: modelString, + ResponseModel: resp.Model, + UseCase: req.UseCase, + ContextScope: assistantContextScopeForExecuteRequest(req), + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + CacheCreationInputTokens: resp.CacheCreationInputTokens, + CacheReadInputTokens: resp.CacheReadInputTokens, + TargetType: req.TargetType, + TargetID: req.TargetID, + FindingID: req.FindingID, }) } @@ -2654,6 +2678,8 @@ Always execute the commands rather than telling the user how to do it.` totalInputTokens += resp.InputTokens totalOutputTokens += resp.OutputTokens + totalCacheCreationTokens += resp.CacheCreationInputTokens + totalCacheReadTokens += resp.CacheReadInputTokens model = resp.Model finalContent = resp.Content @@ -2877,11 +2903,13 @@ Always execute the commands rather than telling the user how to do it.` callback(StreamEvent{Type: "done"}) return &ExecuteResponse{ - Content: finalContent, - Model: model, - InputTokens: totalInputTokens, - OutputTokens: totalOutputTokens, - ToolCalls: toolExecutions, + Content: finalContent, + Model: model, + InputTokens: totalInputTokens, + OutputTokens: totalOutputTokens, + CacheCreationInputTokens: totalCacheCreationTokens, + CacheReadInputTokens: totalCacheReadTokens, + ToolCalls: toolExecutions, }, nil } @@ -3747,11 +3775,13 @@ func (s *Service) AnalyzeForDiscovery(ctx context.Context, prompt string) (strin providerName, _ = config.ParseModelString(model) } costStore.Record(cost.UsageEvent{ - Provider: providerName, - RequestModel: model, - UseCase: "discovery", - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, + Provider: providerName, + RequestModel: model, + UseCase: "discovery", + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + CacheCreationInputTokens: resp.CacheCreationInputTokens, + CacheReadInputTokens: resp.CacheReadInputTokens, }) } diff --git a/internal/api/ai_handlers.go b/internal/api/ai_handlers.go index 0b367402dc..62ee996eff 100644 --- a/internal/api/ai_handlers.go +++ b/internal/api/ai_handlers.go @@ -7265,7 +7265,7 @@ func (h *AISettingsHandler) HandleExportAICostHistory(w http.ResponseWriter, r * exported := make([]exportEvent, 0, len(events)) for _, e := range events { provider, model := cost.ResolveProviderAndModel(e.Provider, e.RequestModel, e.ResponseModel) - usd, ok, _ := cost.EstimateUSD(provider, model, int64(e.InputTokens), int64(e.OutputTokens)) + usd, ok, _ := cost.EstimateUsageUSD(provider, model, e.TokenUsage()) exported = append(exported, exportEvent{ UsageEvent: e, EstimatedUSD: usd, @@ -7292,6 +7292,8 @@ func (h *AISettingsHandler) HandleExportAICostHistory(w http.ResponseWriter, r * "use_case", "input_tokens", "output_tokens", + "cache_creation_input_tokens", + "cache_read_input_tokens", "estimated_usd", "pricing_known", "target_type", @@ -7300,7 +7302,7 @@ func (h *AISettingsHandler) HandleExportAICostHistory(w http.ResponseWriter, r * }) for _, e := range events { provider, model := cost.ResolveProviderAndModel(e.Provider, e.RequestModel, e.ResponseModel) - usd, ok, _ := cost.EstimateUSD(provider, model, int64(e.InputTokens), int64(e.OutputTokens)) + usd, ok, _ := cost.EstimateUsageUSD(provider, model, e.TokenUsage()) _ = cw.Write([]string{ e.Timestamp.UTC().Format(time.RFC3339Nano), @@ -7310,6 +7312,8 @@ func (h *AISettingsHandler) HandleExportAICostHistory(w http.ResponseWriter, r * e.UseCase, strconv.Itoa(e.InputTokens), strconv.Itoa(e.OutputTokens), + strconv.Itoa(e.CacheCreationInputTokens), + strconv.Itoa(e.CacheReadInputTokens), strconv.FormatFloat(usd, 'f', 6, 64), strconv.FormatBool(ok), e.TargetType, diff --git a/internal/api/ai_handlers_more_test.go b/internal/api/ai_handlers_more_test.go index 12bc7899fc..f1d8973d81 100644 --- a/internal/api/ai_handlers_more_test.go +++ b/internal/api/ai_handlers_more_test.go @@ -14,6 +14,7 @@ import ( "github.com/gorilla/websocket" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" "github.com/rcourtman/pulse-go-rewrite/internal/ai" + "github.com/rcourtman/pulse-go-rewrite/internal/ai/cost" "github.com/rcourtman/pulse-go-rewrite/internal/ai/memory" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" @@ -719,3 +720,74 @@ func TestBuildPatrolDigestReportsAvailability(t *testing.T) { t.Fatal("digest must report available once a Patrol service exists") } } + +// Pulse #2350: the usage export must carry the prompt-cache buckets and price +// them at the provider's cache rates, in both formats. +func TestHandleExportAICostHistoryCarriesPromptCacheBuckets(t *testing.T) { + t.Parallel() + + tmp := t.TempDir() + cfg := &config.Config{DataPath: tmp} + persistence := config.NewConfigPersistence(tmp) + handler := newTestAISettingsHandler(cfg, persistence, nil) + store := handler.defaultAIService.CostStore() + if store == nil { + t.Fatal("expected the AI service to own a cost store") + } + store.Record(cost.UsageEvent{ + Timestamp: time.Now(), + Provider: "anthropic", + RequestModel: "anthropic:claude-sonnet-5", + UseCase: "patrol", + InputTokens: 100_000, + OutputTokens: 10_000, + CacheCreationInputTokens: 200_000, + CacheReadInputTokens: 1_000_000, + }) + // 0.1M*2.00 + 0.01M*10.00 + 0.2M*2.50 + 1.0M*0.20 at Anthropic cache rates. + const wantUSD = 1.0 + + req := newLoopbackRequest(http.MethodGet, "/api/ai/cost/export?days=7&format=json", nil) + rec := httptest.NewRecorder() + handler.HandleExportAICostHistory(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("json export status = %d: %s", rec.Code, rec.Body.String()) + } + var exported struct { + Events []struct { + cost.UsageEvent + EstimatedUSD float64 `json:"estimated_usd"` + PricingKnown bool `json:"pricing_known"` + } `json:"events"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &exported); err != nil { + t.Fatalf("decode json export: %v", err) + } + if len(exported.Events) != 1 { + t.Fatalf("expected one exported event, got %d", len(exported.Events)) + } + got := exported.Events[0] + if got.CacheCreationInputTokens != 200_000 || got.CacheReadInputTokens != 1_000_000 || got.InputTokens != 100_000 { + t.Fatalf("json export lost cache buckets: %+v", got.UsageEvent) + } + if !got.PricingKnown || got.EstimatedUSD < wantUSD-1e-6 || got.EstimatedUSD > wantUSD+1e-6 { + t.Fatalf("json export estimated_usd = %f (known=%v), want %f", got.EstimatedUSD, got.PricingKnown, wantUSD) + } + + req = newLoopbackRequest(http.MethodGet, "/api/ai/cost/export?days=7&format=csv", nil) + rec = httptest.NewRecorder() + handler.HandleExportAICostHistory(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("csv export status = %d: %s", rec.Code, rec.Body.String()) + } + lines := strings.Split(strings.TrimSpace(rec.Body.String()), "\n") + if len(lines) != 2 { + t.Fatalf("expected header and one row, got %d lines: %q", len(lines), rec.Body.String()) + } + if !strings.Contains(lines[0], "input_tokens,output_tokens,cache_creation_input_tokens,cache_read_input_tokens,estimated_usd,") { + t.Fatalf("csv header missing cache columns after output_tokens: %s", lines[0]) + } + if !strings.Contains(lines[1], ",100000,10000,200000,1000000,1.000000,true,") { + t.Fatalf("csv row missing cache buckets or cache-rate pricing: %s", lines[1]) + } +} diff --git a/internal/api/chat_service_adapter.go b/internal/api/chat_service_adapter.go index e84b647df5..8870e97752 100644 --- a/internal/api/chat_service_adapter.go +++ b/internal/api/chat_service_adapter.go @@ -38,17 +38,21 @@ func (a *chatServiceAdapter) ExecutePatrolStream(ctx context.Context, req ai.Pat if err != nil { if resp != nil { return &ai.PatrolStreamResponse{ - Content: resp.Content, - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, + Content: resp.Content, + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + CacheCreationInputTokens: resp.CacheCreationInputTokens, + CacheReadInputTokens: resp.CacheReadInputTokens, }, err } return nil, err } return &ai.PatrolStreamResponse{ - Content: resp.Content, - InputTokens: resp.InputTokens, - OutputTokens: resp.OutputTokens, + Content: resp.Content, + InputTokens: resp.InputTokens, + OutputTokens: resp.OutputTokens, + CacheCreationInputTokens: resp.CacheCreationInputTokens, + CacheReadInputTokens: resp.CacheReadInputTokens, }, nil } diff --git a/internal/config/persistence.go b/internal/config/persistence.go index 17d5eb11cf..c1eefbf941 100644 --- a/internal/config/persistence.go +++ b/internal/config/persistence.go @@ -2969,9 +2969,11 @@ type PatrolRunRecord struct { TriageFlags int `json:"triage_flags"` TriageSkippedLLM bool `json:"triage_skipped_llm,omitempty"` // AI Analysis details - AIAnalysis string `json:"ai_analysis,omitempty"` // The AI's raw response/analysis - InputTokens int `json:"input_tokens,omitempty"` // Tokens sent to AI - OutputTokens int `json:"output_tokens,omitempty"` // Tokens received from AI + AIAnalysis string `json:"ai_analysis,omitempty"` // The AI's raw response/analysis + InputTokens int `json:"input_tokens,omitempty"` // Tokens sent to AI + OutputTokens int `json:"output_tokens,omitempty"` // Tokens received from AI + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` // Prompt-cache writes, disjoint from InputTokens + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` // Prompt-cache reads, disjoint from InputTokens // Tool call traces ToolCalls []ToolCallRecord `json:"tool_calls,omitempty"` ToolCallCount int `json:"tool_call_count"` @@ -3017,6 +3019,8 @@ type patrolRunRecordJSON struct { AIAnalysis string `json:"ai_analysis,omitempty"` InputTokens int `json:"input_tokens,omitempty"` OutputTokens int `json:"output_tokens,omitempty"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` // Prompt-cache writes, disjoint from InputTokens + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` // Prompt-cache reads, disjoint from InputTokens ToolCalls []ToolCallRecord `json:"tool_calls,omitempty"` ToolCallCount int `json:"tool_call_count"` } @@ -3108,6 +3112,8 @@ func (r PatrolRunRecord) MarshalJSON() ([]byte, error) { AIAnalysis: normalized.AIAnalysis, InputTokens: normalized.InputTokens, OutputTokens: normalized.OutputTokens, + CacheCreationInputTokens: normalized.CacheCreationInputTokens, + CacheReadInputTokens: normalized.CacheReadInputTokens, ToolCalls: normalized.ToolCalls, ToolCallCount: normalized.ToolCallCount, }) @@ -3160,6 +3166,8 @@ func (r *PatrolRunRecord) UnmarshalJSON(data []byte) error { AIAnalysis: payload.AIAnalysis, InputTokens: payload.InputTokens, OutputTokens: payload.OutputTokens, + CacheCreationInputTokens: payload.CacheCreationInputTokens, + CacheReadInputTokens: payload.CacheReadInputTokens, ToolCalls: payload.ToolCalls, ToolCallCount: payload.ToolCallCount, } @@ -3188,18 +3196,20 @@ type AIUsageHistoryData struct { // AIUsageEventRecord is a persisted usage event for an AI provider call. // This intentionally excludes prompt/response content for privacy. type AIUsageEventRecord struct { - Timestamp time.Time `json:"timestamp"` - Provider string `json:"provider"` - RequestModel string `json:"request_model"` - ResponseModel string `json:"response_model,omitempty"` - UseCase string `json:"use_case,omitempty"` // "chat" or "patrol" - ContextScope string `json:"context_scope,omitempty"` - ToolCallCount int `json:"tool_call_count,omitempty"` - InputTokens int `json:"input_tokens,omitempty"` - OutputTokens int `json:"output_tokens,omitempty"` - TargetType string `json:"target_type,omitempty"` - TargetID string `json:"target_id,omitempty"` - FindingID string `json:"finding_id,omitempty"` + Timestamp time.Time `json:"timestamp"` + Provider string `json:"provider"` + RequestModel string `json:"request_model"` + ResponseModel string `json:"response_model,omitempty"` + UseCase string `json:"use_case,omitempty"` // "chat" or "patrol" + ContextScope string `json:"context_scope,omitempty"` + ToolCallCount int `json:"tool_call_count,omitempty"` + InputTokens int `json:"input_tokens,omitempty"` + OutputTokens int `json:"output_tokens,omitempty"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` // Prompt-cache writes, disjoint from InputTokens + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` // Prompt-cache reads, disjoint from InputTokens + TargetType string `json:"target_type,omitempty"` + TargetID string `json:"target_id,omitempty"` + FindingID string `json:"finding_id,omitempty"` } const (