diff --git a/desktop/tabs.go b/desktop/tabs.go index e00487b673..5c1751a210 100644 --- a/desktop/tabs.go +++ b/desktop/tabs.go @@ -247,13 +247,17 @@ type readFileRecord struct { } type sessionUsageStats struct { - PromptTokens int `json:"promptTokens"` - CompletionTokens int `json:"completionTokens"` - TotalTokens int `json:"totalTokens"` - ReasoningTokens int `json:"reasoningTokens"` - CacheHitTokens int `json:"cacheHitTokens"` - CacheMissTokens int `json:"cacheMissTokens"` - Estimated bool `json:"estimated,omitempty"` + PromptTokens int `json:"promptTokens"` + CompletionTokens int `json:"completionTokens"` + TotalTokens int `json:"totalTokens"` + ReasoningTokens int `json:"reasoningTokens"` + CacheHitTokens int `json:"cacheHitTokens"` + CacheMissTokens int `json:"cacheMissTokens"` + CacheWriteTokens int `json:"cacheWriteTokens,omitempty"` + // CacheWriteBilledTokens preserves provider-specific cache-write pricing + // across persisted telemetry repricing without changing hit-rate totals. + CacheWriteBilledTokens float64 `json:"cacheWriteBilledTokens,omitempty"` + Estimated bool `json:"estimated,omitempty"` // LastUsedTokens is the executor-reported context fill (prompt+completion) // from the most recent turn. It is persisted so the status bar / context // panel can show a meaningful fill percentage after a session rebind @@ -280,17 +284,19 @@ type sessionUsageStats struct { } type usageSourceStats struct { - PromptTokens int `json:"promptTokens"` - CompletionTokens int `json:"completionTokens"` - TotalTokens int `json:"totalTokens"` - ReasoningTokens int `json:"reasoningTokens"` - CacheHitTokens int `json:"cacheHitTokens"` - CacheMissTokens int `json:"cacheMissTokens"` - Estimated bool `json:"estimated,omitempty"` - RequestCount int `json:"requestCount"` - SessionCost float64 `json:"sessionCost,omitempty"` - SessionCurrency string `json:"sessionCurrency,omitempty"` - SessionCostUsd float64 `json:"sessionCostUsd,omitempty"` + PromptTokens int `json:"promptTokens"` + CompletionTokens int `json:"completionTokens"` + TotalTokens int `json:"totalTokens"` + ReasoningTokens int `json:"reasoningTokens"` + CacheHitTokens int `json:"cacheHitTokens"` + CacheMissTokens int `json:"cacheMissTokens"` + CacheWriteTokens int `json:"cacheWriteTokens,omitempty"` + CacheWriteBilledTokens float64 `json:"cacheWriteBilledTokens,omitempty"` + Estimated bool `json:"estimated,omitempty"` + RequestCount int `json:"requestCount"` + SessionCost float64 `json:"sessionCost,omitempty"` + SessionCurrency string `json:"sessionCurrency,omitempty"` + SessionCostUsd float64 `json:"sessionCostUsd,omitempty"` } type sourceSessionCacheCounters struct { @@ -966,6 +972,8 @@ func (t *WorkspaceTab) recordUsage(e event.Event) { cacheHitTokens, cacheMissTokens := t.usageTelemetry.cacheTokenDelta(source, u, e.SessionHit, e.SessionMiss) t.usageTelemetry.CacheHitTokens += cacheHitTokens t.usageTelemetry.CacheMissTokens += cacheMissTokens + t.usageTelemetry.CacheWriteTokens += u.CacheWriteTokens + t.usageTelemetry.CacheWriteBilledTokens += u.CacheWriteBilledTokens t.usageTelemetry.Estimated = t.usageTelemetry.Estimated || u.Estimated requestCount := u.RequestCount if requestCount <= 0 { @@ -996,6 +1004,8 @@ func (t *WorkspaceTab) recordUsage(e event.Event) { src.ReasoningTokens += u.ReasoningTokens src.CacheHitTokens += cacheHitTokens src.CacheMissTokens += cacheMissTokens + src.CacheWriteTokens += u.CacheWriteTokens + src.CacheWriteBilledTokens += u.CacheWriteBilledTokens src.Estimated = src.Estimated || u.Estimated src.RequestCount += requestCount if e.Pricing != nil { @@ -1029,25 +1039,29 @@ func (t *WorkspaceTab) recordUsage(e event.Event) { func usageStatsAsProviderUsage(stats usageSourceStats) *provider.Usage { return &provider.Usage{ - PromptTokens: stats.PromptTokens, - CompletionTokens: stats.CompletionTokens, - TotalTokens: stats.TotalTokens, - ReasoningTokens: stats.ReasoningTokens, - CacheHitTokens: stats.CacheHitTokens, - CacheMissTokens: stats.CacheMissTokens, - Estimated: stats.Estimated, + PromptTokens: stats.PromptTokens, + CompletionTokens: stats.CompletionTokens, + TotalTokens: stats.TotalTokens, + ReasoningTokens: stats.ReasoningTokens, + CacheHitTokens: stats.CacheHitTokens, + CacheMissTokens: stats.CacheMissTokens, + CacheWriteTokens: stats.CacheWriteTokens, + CacheWriteBilledTokens: stats.CacheWriteBilledTokens, + Estimated: stats.Estimated, } } func sessionStatsAsProviderUsage(stats sessionUsageStats) *provider.Usage { return &provider.Usage{ - PromptTokens: stats.PromptTokens, - CompletionTokens: stats.CompletionTokens, - TotalTokens: stats.TotalTokens, - ReasoningTokens: stats.ReasoningTokens, - CacheHitTokens: stats.CacheHitTokens, - CacheMissTokens: stats.CacheMissTokens, - Estimated: stats.Estimated, + PromptTokens: stats.PromptTokens, + CompletionTokens: stats.CompletionTokens, + TotalTokens: stats.TotalTokens, + ReasoningTokens: stats.ReasoningTokens, + CacheHitTokens: stats.CacheHitTokens, + CacheMissTokens: stats.CacheMissTokens, + CacheWriteTokens: stats.CacheWriteTokens, + CacheWriteBilledTokens: stats.CacheWriteBilledTokens, + Estimated: stats.Estimated, } } diff --git a/desktop/tabs_telemetry_test.go b/desktop/tabs_telemetry_test.go index dc708c5028..abb504225c 100644 --- a/desktop/tabs_telemetry_test.go +++ b/desktop/tabs_telemetry_test.go @@ -135,6 +135,34 @@ func TestWorkspaceTabRepricesUsageWithoutMixingCurrencies(t *testing.T) { } } +func TestWorkspaceTabRepricesCacheWritesWithoutLosingBillingTier(t *testing.T) { + tab := &WorkspaceTab{} + tab.recordUsage(event.Event{ + Usage: &provider.Usage{ + PromptTokens: 500_000, + TotalTokens: 500_000, + CacheMissTokens: 500_000, + CacheWriteTokens: 100_000, + CacheWriteBilledTokens: 200_000, + }, + UsageSource: event.UsageSourceExecutor, + Pricing: &provider.Pricing{Input: 2, Currency: "CNY"}, + }) + if ok := tab.repriceUsage(map[string]*provider.Pricing{ + event.UsageSourceExecutor: {Input: 1, Currency: "USD"}, + }); !ok { + t.Fatal("repriceUsage rejected cache-write usage") + } + got := tab.telemetrySnapshot().Usage + // 400K ordinary input units + 200K billed cache-write units. + if got.SessionCurrency != "$" || got.SessionCost != 0.6 { + t.Fatalf("repriced cache-write usage = %f %q, want 0.6 USD", got.SessionCost, got.SessionCurrency) + } + if got.CacheWriteTokens != 100_000 || got.CacheWriteBilledTokens != 200_000 { + t.Fatalf("persisted cache writes = raw %d billed %v", got.CacheWriteTokens, got.CacheWriteBilledTokens) + } +} + func TestRepriceTabUsageUsesDetectedLocaleForAutoCurrency(t *testing.T) { isolateDesktopUserDirs(t) cfg := config.Default() @@ -278,14 +306,16 @@ func TestTelemetryLastContextRoundTripAndLegacyDefaults(t *testing.T) { want := tabTelemetrySnapshot{ Version: 2, Usage: sessionUsageStats{ - PromptTokens: 100, - TotalTokens: 120, - LastUsedTokens: 120, - LastPromptTokens: 100, - LastCompletionTokens: 20, - LastReasoningTokens: 8, - LastCacheHitTokens: 70, - LastCacheMissTokens: 30, + PromptTokens: 100, + TotalTokens: 120, + CacheWriteTokens: 5, + CacheWriteBilledTokens: 10, + LastUsedTokens: 120, + LastPromptTokens: 100, + LastCompletionTokens: 20, + LastReasoningTokens: 8, + LastCacheHitTokens: 70, + LastCacheMissTokens: 30, }, } if err := saveTelemetry(path, want); err != nil { @@ -300,11 +330,17 @@ func TestTelemetryLastContextRoundTripAndLegacyDefaults(t *testing.T) { got.LastCacheMissTokens != want.Usage.LastCacheMissTokens { t.Fatalf("last context round trip = %+v, want %+v", got, want.Usage) } + if got.CacheWriteTokens != 5 || got.CacheWriteBilledTokens != 10 { + t.Fatalf("cache-write round trip = raw %d billed %v, want 5/10", got.CacheWriteTokens, got.CacheWriteBilledTokens) + } if err := os.WriteFile(path, []byte(`{"version":2,"usage":{"promptTokens":50,"totalTokens":50}}`), 0o644); err != nil { t.Fatalf("write pre-last-context telemetry: %v", err) } legacy := loadTelemetry(path).Usage + if legacy.CacheWriteTokens != 0 || legacy.CacheWriteBilledTokens != 0 { + t.Fatalf("legacy cache-write fields = raw %d billed %v, want zero defaults", legacy.CacheWriteTokens, legacy.CacheWriteBilledTokens) + } if legacy.LastUsedTokens != 0 || legacy.LastPromptTokens != 0 || legacy.LastCompletionTokens != 0 || diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 27a678576a..9614f12b47 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -286,13 +286,6 @@ type Agent struct { // run loop writes it while a frontend's status line reads it, so it is atomic. lastUsage atomic.Pointer[provider.Usage] - // lastRequestSentAt is the wall-clock moment (unix millis) the most recent - // provider attempt was sent. Each attempt updates it, while all attempts in - // one frozen sampling round reuse the first attempt's gap hint so adaptive - // cache TTL cannot change the provider-visible replay body. Zero means no - // request sent yet — a fresh process or sub-agent gets the cheaper default TTL. - lastRequestSentAt atomic.Int64 - // sessCacheHit/sessCacheMiss accumulate cache tokens across every API call // this session, so frontends can show the aggregate hit-rate (Σhit/Σ(hit+miss)) // — a steadier, cost-oriented number than the single-turn rate. They are NOT @@ -2876,10 +2869,7 @@ func toolBudgetNoticeText() string { // round. All stream retries replay this exact payload — no synthetic recovery // messages, no schema reorder, no previous_response_id drift from failed attempts. type samplingRequest struct { - req provider.Request - cacheGap time.Duration - cacheGapKnown bool - cacheGapPresent bool + req provider.Request } // prepareSamplingRequest runs interceptors and schema fetch once per model @@ -2996,30 +2986,9 @@ func (a *Agent) streamWithFrozen(ctx context.Context, turn int, sink event.Sink, } req = prepared.req } - // Stamp every actual attempt at send time, but freeze the first attempt's - // cache-gap hint for the entire sampling lifecycle. Anthropic derives an - // ephemeral-cache TTL from this context value, so recomputing it on an - // immediate retry would change the HTTP body despite an identical Request. - sentAt := time.Now() - priorMs := a.lastRequestSentAt.Swap(sentAt.UnixMilli()) - if frozen == nil { - if priorMs != 0 { - ctx = provider.WithCacheGapHint(ctx, sentAt.Sub(time.UnixMilli(priorMs))) - } - } else { - if !frozen.cacheGapKnown { - frozen.cacheGapKnown = true - if priorMs != 0 { - frozen.cacheGap = sentAt.Sub(time.UnixMilli(priorMs)) - frozen.cacheGapPresent = true - } - } - if frozen.cacheGapPresent { - ctx = provider.WithCacheGapHint(ctx, frozen.cacheGap) - } - } - // After #7725 Goal token request admission was removed; stream goes - // directly to the provider while still carrying the cache-gap hint. + // After #7725 Goal token request admission was removed, stream goes + // directly to the provider. Provider-visible cache controls stay stable + // across retries and request timing because they are derived from req alone. ch, err := a.prov.Stream(ctx, req) if err != nil { return "", "", "", "", "", nil, nil, provider.UsageWithRequestAttemptCount(ctx, nil), false, false, nil, 0, err diff --git a/internal/agent/cache_gap_hint_test.go b/internal/agent/cache_gap_hint_test.go deleted file mode 100644 index dca35cffb0..0000000000 --- a/internal/agent/cache_gap_hint_test.go +++ /dev/null @@ -1,84 +0,0 @@ -package agent - -import ( - "context" - "testing" - "time" - - "reasonix/internal/event" - "reasonix/internal/provider" - "reasonix/internal/tool" -) - -// gapCaptureProvider records whether/what cache-gap hint arrived on each -// Stream call, so tests can prove Agent.stream threads it into Provider.Stream -// and freezes it across body retries. -type gapCaptureProvider struct { - gaps []time.Duration - ok []bool - interruptNext bool -} - -func (p *gapCaptureProvider) Name() string { return "gap-capture" } - -func (p *gapCaptureProvider) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Chunk, error) { - gap, ok := provider.CacheGapHintFromContext(ctx) - p.gaps = append(p.gaps, gap) - p.ok = append(p.ok, ok) - ch := make(chan provider.Chunk, 2) - if p.interruptNext { - p.interruptNext = false - ch <- provider.Chunk{Type: provider.ChunkError, Err: provider.StreamInterrupt(context.Canceled, provider.StreamInterruptConnectionReset)} - close(ch) - return ch, nil - } - ch <- provider.Chunk{Type: provider.ChunkText, Text: "ok"} - ch <- provider.Chunk{Type: provider.ChunkUsage, Usage: &provider.Usage{PromptTokens: 10, CompletionTokens: 1, TotalTokens: 11}} - close(ch) - return ch, nil -} - -func TestStreamRecoveryFreezesCacheGapHint(t *testing.T) { - prov := &gapCaptureProvider{} - a := New(prov, tool.NewRegistry(), NewSession("sys"), Options{}, event.Discard) - if err := a.Run(context.Background(), "prime"); err != nil { - t.Fatalf("prime Run: %v", err) - } - time.Sleep(5 * time.Millisecond) - prov.interruptNext = true - if err := a.Run(context.Background(), "retry"); err != nil { - t.Fatalf("recovered Run: %v", err) - } - - if len(prov.gaps) != 3 { - t.Fatalf("Stream calls = %d, want prime + failed attempt + retry", len(prov.gaps)) - } - if !prov.ok[1] || !prov.ok[2] || prov.gaps[1] <= 0 { - t.Fatalf("retry round hints = ok %v/%v gaps %v/%v", prov.ok[1], prov.ok[2], prov.gaps[1], prov.gaps[2]) - } - if prov.gaps[2] != prov.gaps[1] { - t.Fatalf("retry cache gap changed provider-visible TTL input: first=%v retry=%v", prov.gaps[1], prov.gaps[2]) - } -} - -func TestStreamCarriesCacheGapHintAfterFirstTurn(t *testing.T) { - prov := &gapCaptureProvider{} - a := New(prov, tool.NewRegistry(), NewSession("sys"), Options{}, event.Discard) - if err := a.Run(context.Background(), "one"); err != nil { - t.Fatalf("first Run: %v", err) - } - time.Sleep(5 * time.Millisecond) - if err := a.Run(context.Background(), "two"); err != nil { - t.Fatalf("second Run: %v", err) - } - - if len(prov.ok) != 2 { - t.Fatalf("got %d Stream calls, want 2", len(prov.ok)) - } - if prov.ok[0] { - t.Fatalf("first request should carry no gap hint, got gap=%v", prov.gaps[0]) - } - if !prov.ok[1] || prov.gaps[1] <= 0 { - t.Fatalf("second request should carry a positive gap hint, got ok=%v gap=%v", prov.ok[1], prov.gaps[1]) - } -} diff --git a/internal/agent/run_loop.go b/internal/agent/run_loop.go index cd0ab06940..82593f167b 100644 --- a/internal/agent/run_loop.go +++ b/internal/agent/run_loop.go @@ -719,6 +719,8 @@ func mergeSamplingUsage(acc, attempt *provider.Usage) *provider.Usage { } merged.CacheHitTokens = ah + bh merged.CacheMissTokens = am + bm + merged.CacheWriteTokens += attempt.CacheWriteTokens + merged.CacheWriteBilledTokens += attempt.CacheWriteBilledTokens merged.PromptTokens = billablePrompt(merged.CacheHitTokens, merged.CacheMissTokens, 0) if merged.PromptTokens == 0 { merged.PromptTokens = acc.PromptTokens + attempt.PromptTokens diff --git a/internal/agent/usage_accounting_test.go b/internal/agent/usage_accounting_test.go index 85367c3036..a8e052af28 100644 --- a/internal/agent/usage_accounting_test.go +++ b/internal/agent/usage_accounting_test.go @@ -39,8 +39,8 @@ func (failedRequestProvider) Stream(ctx context.Context, _ provider.Request) (<- } func TestMergeStreamUsageCountsProviderRequests(t *testing.T) { - first := &provider.Usage{PromptTokens: 10, CompletionTokens: 5, TotalTokens: 15, RequestCount: 1} - retry := &provider.Usage{PromptTokens: 20, CompletionTokens: 8, TotalTokens: 28, RequestCount: 1} + first := &provider.Usage{PromptTokens: 10, CompletionTokens: 5, TotalTokens: 15, CacheWriteTokens: 2, CacheWriteBilledTokens: 2.5, RequestCount: 1} + retry := &provider.Usage{PromptTokens: 20, CompletionTokens: 8, TotalTokens: 28, CacheWriteTokens: 3, CacheWriteBilledTokens: 6, RequestCount: 1} got := mergeStreamUsage(first, retry) if got == nil || got.TotalTokens != 43 || got.RequestCount != 2 || got.CompletionTokens != 13 { t.Fatalf("merged usage = %+v, want total=43 requests=2 completion=13", got) @@ -49,6 +49,9 @@ func TestMergeStreamUsageCountsProviderRequests(t *testing.T) { if got.CacheMissTokens != 30 || got.PromptTokens != 30 { t.Fatalf("billable input = prompt %d miss %d, want 30/30", got.PromptTokens, got.CacheMissTokens) } + if got.CacheWriteTokens != 5 || got.CacheWriteBilledTokens != 8.5 { + t.Fatalf("merged cache writes = raw %d billed %v, want 5/8.5", got.CacheWriteTokens, got.CacheWriteBilledTokens) + } third := &provider.Usage{PromptTokens: 1, CompletionTokens: 1, TotalTokens: 2, RequestCount: 1} got = mergeStreamUsage(got, third) diff --git a/internal/cli/run_metrics.go b/internal/cli/run_metrics.go index 4eceb5fa84..2732f45600 100644 --- a/internal/cli/run_metrics.go +++ b/internal/cli/run_metrics.go @@ -142,6 +142,7 @@ func (s *metricsSink) Snapshot() RunMetrics { func (m RunMetrics) clone() RunMetrics { out := m + out.PrefixChangeReasonCounts = cloneCounts(m.PrefixChangeReasonCounts) out.UsageBySource = cloneSourceUsage(m.UsageBySource) out.ToolCallsByName = cloneCounts(m.ToolCallsByName) out.ToolFailuresByName = cloneCounts(m.ToolFailuresByName) @@ -208,9 +209,7 @@ func (s *metricsSink) record(e event.Event) { s.m.Estimated = s.m.Estimated || u.Estimated var stepCost float64 if p := e.Pricing; p != nil { - stepCost = (float64(u.CacheHitTokens)*p.CacheHit + - float64(u.CacheMissTokens)*p.Input + - float64(u.CompletionTokens)*p.Output) / 1e6 + stepCost = p.Cost(u) s.m.Cost += stepCost s.m.Currency = p.Currency } diff --git a/internal/cli/run_metrics_partial_test.go b/internal/cli/run_metrics_partial_test.go index 26ced170ac..a2b7194d18 100644 --- a/internal/cli/run_metrics_partial_test.go +++ b/internal/cli/run_metrics_partial_test.go @@ -21,6 +21,27 @@ func usageEvent(source string, prompt, completion int) event.Event { } } +func usageEventWithCacheReason(reason string) event.Event { + e := usageEvent(event.UsageSourceSubagent, 10, 1) + e.CacheDiagnostics = &event.CacheDiagnostics{PrefixChangeReasons: []string{reason}} + return e +} + +func TestSnapshotDeepCopiesPrefixChangeReasons(t *testing.T) { + s := &metricsSink{inner: event.Discard} + s.Emit(usageEventWithCacheReason("compact_auto")) + + snapshot := s.Snapshot() + s.Emit(usageEventWithCacheReason("snip")) + + if snapshot.PrefixChangeReasonCounts["compact_auto"] != 1 { + t.Fatalf("snapshot compact_auto = %d, want 1", snapshot.PrefixChangeReasonCounts["compact_auto"]) + } + if _, changed := snapshot.PrefixChangeReasonCounts["snip"]; changed { + t.Fatalf("snapshot changed after return: %v", snapshot.PrefixChangeReasonCounts) + } +} + // A killed agent writes no final record. Everything it did before the kill is // only recoverable if snapshots landed on disk while it ran. func TestSnapshotSurvivesWithoutAFinalWrite(t *testing.T) { @@ -173,7 +194,7 @@ func TestConcurrentEmitAndSnapshotAreRaceFree(t *testing.T) { go func() { defer wg.Done() for j := 0; j < each; j++ { - s.Emit(usageEvent(event.UsageSourceSubagent, 10, 1)) + s.Emit(usageEventWithCacheReason("compact_auto")) s.Emit(event.Event{Kind: event.ToolResult, Tool: event.Tool{Name: "bash"}}) } }() @@ -182,7 +203,10 @@ func TestConcurrentEmitAndSnapshotAreRaceFree(t *testing.T) { go func() { defer wg.Done() for i := 0; i < 200; i++ { - _ = s.Snapshot() + if _, err := json.Marshal(s.Snapshot()); err != nil { + t.Errorf("marshal snapshot: %v", err) + return + } } }() wg.Wait() diff --git a/internal/cli/run_metrics_test.go b/internal/cli/run_metrics_test.go index aec375a790..572356998b 100644 --- a/internal/cli/run_metrics_test.go +++ b/internal/cli/run_metrics_test.go @@ -8,8 +8,25 @@ import ( "reasonix/internal/event" "reasonix/internal/evidence" + "reasonix/internal/provider" ) +func TestMetricsSinkUsesProviderCacheWriteCost(t *testing.T) { + s := &metricsSink{inner: event.Discard} + s.Emit(event.Event{ + Kind: event.Usage, + Usage: &provider.Usage{ + CacheMissTokens: 500_000, + CacheWriteTokens: 100_000, + CacheWriteBilledTokens: 200_000, + }, + Pricing: &provider.Pricing{Input: 2}, + }) + if got := s.Snapshot().Cost; got != 1.2 { + t.Fatalf("metrics cost = %f, want 1.2", got) + } +} + func TestMetricsSinkAccumulatesReadinessAudit(t *testing.T) { s := &metricsSink{inner: event.Discard} diff --git a/internal/guardian/guardian.go b/internal/guardian/guardian.go index cbc11d6494..fef31113bf 100644 --- a/internal/guardian/guardian.go +++ b/internal/guardian/guardian.go @@ -593,6 +593,8 @@ func (gs *Session) addReviewUsage(usage *provider.Usage) { gs.reviewUsage.TotalTokens += usage.TotalTokens gs.reviewUsage.CacheHitTokens += usage.CacheHitTokens gs.reviewUsage.CacheMissTokens += usage.CacheMissTokens + gs.reviewUsage.CacheWriteTokens += usage.CacheWriteTokens + gs.reviewUsage.CacheWriteBilledTokens += usage.CacheWriteBilledTokens gs.reviewUsage.ReasoningTokens += usage.ReasoningTokens gs.reviewUsage.RequestCount += guardianUsageRequestCount(usage) gs.reviewUsage.Estimated = gs.reviewUsage.Estimated || usage.Estimated diff --git a/internal/provider/anthropic/anthropic.go b/internal/provider/anthropic/anthropic.go index 144b8ec4cc..aed93dd639 100644 --- a/internal/provider/anthropic/anthropic.go +++ b/internal/provider/anthropic/anthropic.go @@ -116,6 +116,7 @@ func New(cfg provider.Config) (provider.Provider, error) { keySource: keySource, baseURL: root, model: cfg.Model, + nativeAnthropic: strings.EqualFold(root, defaultBaseURL), deepseek: openai.IsDeepSeek(root), thinking: thinking, effort: effort, @@ -142,6 +143,7 @@ type client struct { keySource string // source of keyEnv, surfaced in auth errors baseURL string model string + nativeAnthropic bool // first-party endpoint: documented default-5m cache-write pricing applies deepseek bool // official DeepSeek Anthropic endpoint: unsigned reasoning replay + automatic cache thinking string // "adaptive" enables extended thinking; "" = off (config-driven) effort string // output_config.effort: low|medium|high|xhigh|max; "" = provider default @@ -302,7 +304,7 @@ func (c *client) Stream(ctx context.Context, req provider.Request) (<-chan provi // become `tool_use` blocks; RoleTool results become `tool_result` blocks in a user // turn. Consecutive same-role messages are coalesced because the API requires // alternating user/assistant turns (tool results are user turns). -func (c *client) buildRequest(ctx context.Context, req provider.Request) anthRequest { +func (c *client) buildRequest(_ context.Context, req provider.Request) anthRequest { var system []textBlock var msgs []anthMessage @@ -395,23 +397,19 @@ func (c *client) buildRequest(ctx context.Context, req provider.Request) anthReq // tools → system → messages, so a marker on the last system block caches // tools+system together; with no system, mark the last tool. A marker on the // last block of the last message caches the conversation prefix, accruing hits - // incrementally as turns are appended. Max 4 breakpoints; we use ≤2. One TTL - // decision (default 5m, or 1h when the previous request was long enough ago - // that the 5m window would likely already have expired) applies to all of - // them — this is a single "extend this prefix" choice, not per-block. + // incrementally as turns are appended. Max 4 breakpoints; we use ≤2. Keep + // Anthropic's default 5m TTL by omitting the ttl field. Besides being cheaper + // than the opt-in 1h write, this keeps provider-visible request bytes stable + // across turns, retries, and wall-clock timing. if !c.deepseek { - ttl := "" - if useLongCacheTTL(ctx) { - ttl = cacheTTL1Hour - } if n := len(system); n > 0 { - system[n-1].CacheControl = ephemeral(ttl) + system[n-1].CacheControl = ephemeral() } else if n := len(tools); n > 0 { - tools[n-1].CacheControl = ephemeral(ttl) + tools[n-1].CacheControl = ephemeral() } if n := len(msgs); n > 0 { if k := len(msgs[n-1].Content); k > 0 { - msgs[n-1].Content[k-1].CacheControl = ephemeral(ttl) + msgs[n-1].Content[k-1].CacheControl = ephemeral() } } } @@ -687,13 +685,19 @@ func (c *client) readStream(ctx context.Context, resp *http.Response, out chan<- finalize: if haveUsage { + cacheWriteBilledTokens := 0.0 + if cacheCreate > 0 && c.nativeAnthropic { + cacheWriteBilledTokens = float64(cacheCreate) * cacheWrite5MinuteInputMultiplier + } usage := &provider.Usage{ - PromptTokens: inTok + cacheCreate + cacheRead, - CompletionTokens: outTok, - TotalTokens: inTok + cacheCreate + cacheRead + outTok, - CacheHitTokens: cacheRead, - CacheMissTokens: inTok + cacheCreate, // uncached input + cache writes (billed ≥1×) - FinishReason: mapStopReason(stopReason), + PromptTokens: inTok + cacheCreate + cacheRead, + CompletionTokens: outTok, + TotalTokens: inTok + cacheCreate + cacheRead + outTok, + CacheHitTokens: cacheRead, + CacheMissTokens: inTok + cacheCreate, + CacheWriteTokens: cacheCreate, + CacheWriteBilledTokens: cacheWriteBilledTokens, + FinishReason: mapStopReason(stopReason), } provider.ApplyRequestAttemptCount(ctx, usage) if !send(provider.Chunk{Type: provider.ChunkUsage, Usage: usage}) { @@ -770,33 +774,12 @@ func formatWebSearchResults(raw json.RawMessage) string { // --- Messages API wire protocol --- -// cacheGapLongTTLThreshold is how long the gap since the previous request must -// be before the 1h cache breakpoint (2x write cost) is worth it over the -// default 5m one (1.25x write cost). Comfortably under Anthropic's 5-minute -// expiry so the switch happens before that window would actually go cold, but -// high enough that one ordinary ~60-90s tool round (a build, a test run) -// doesn't flip it: an unneeded 1h choice costs proportionally more (2.1x) than -// an unneeded 5m one (1.35x) would have, so this errs conservative. -const cacheGapLongTTLThreshold = 120 * time.Second +const cacheWrite5MinuteInputMultiplier = 1.25 -// cacheTTL1Hour is the opt-in extended cache_control TTL value (GA, no beta -// header required). An empty TTL string omits the field entirely, which is -// Anthropic's default 5-minute window. -const cacheTTL1Hour = "1h" - -// useLongCacheTTL reports whether the gap since the previous request (see -// provider.WithCacheGapHint) is large enough that the default 5-minute -// prompt-cache window would likely already have expired. -func useLongCacheTTL(ctx context.Context) bool { - gap, ok := provider.CacheGapHintFromContext(ctx) - return ok && gap >= cacheGapLongTTLThreshold -} - -func ephemeral(ttl string) *cacheControl { return &cacheControl{Type: "ephemeral", TTL: ttl} } +func ephemeral() *cacheControl { return &cacheControl{Type: "ephemeral"} } type cacheControl struct { Type string `json:"type"` - TTL string `json:"ttl,omitempty"` } type anthRequest struct { diff --git a/internal/provider/anthropic/anthropic_test.go b/internal/provider/anthropic/anthropic_test.go index aff556c84c..839f8438cd 100644 --- a/internal/provider/anthropic/anthropic_test.go +++ b/internal/provider/anthropic/anthropic_test.go @@ -9,7 +9,6 @@ import ( "net/http/httptest" "strings" "testing" - "time" "reasonix/internal/provider" ) @@ -100,77 +99,35 @@ func TestBuildRequestNoSystem(t *testing.T) { } } -func lastBlockCacheControl(r anthRequest) *cacheControl { - last := r.Messages[len(r.Messages)-1] - return last.Content[len(last.Content)-1].CacheControl -} - -func TestBuildRequestDefaultsToShortCacheTTL(t *testing.T) { +func TestBuildRequestKeepsDefaultCacheControlBytesStable(t *testing.T) { c := &client{model: "claude-opus-4-8"} - r := c.buildRequest(context.Background(), provider.Request{Messages: []provider.Message{{Role: provider.RoleUser, Content: "hi"}}}) - got := lastBlockCacheControl(r) - if got == nil || got.TTL != "" { - t.Fatalf("cache_control = %+v, want empty TTL with no gap hint", got) + req := provider.Request{Messages: []provider.Message{{Role: provider.RoleUser, Content: "hi"}}} + want, err := json.Marshal(c.buildRequest(context.Background(), req)) + if err != nil { + t.Fatalf("marshal first request: %v", err) } -} - -func TestBuildRequestUsesLongCacheTTLOnLargeGap(t *testing.T) { - c := &client{model: "claude-opus-4-8"} - ctx := provider.WithCacheGapHint(context.Background(), 3*time.Minute) - r := c.buildRequest(ctx, provider.Request{Messages: []provider.Message{{Role: provider.RoleUser, Content: "hi"}}}) - got := lastBlockCacheControl(r) - if got == nil || got.TTL != "1h" { - t.Fatalf("cache_control = %+v, want ttl=1h for a 3-minute gap", got) + requestCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + got, err := json.Marshal(c.buildRequest(requestCtx, req)) + if err != nil { + t.Fatalf("marshal second request: %v", err) } -} - -func TestBuildRequestKeepsShortCacheTTLBelowThreshold(t *testing.T) { - c := &client{model: "claude-opus-4-8"} - ctx := provider.WithCacheGapHint(context.Background(), 10*time.Second) - r := c.buildRequest(ctx, provider.Request{Messages: []provider.Message{{Role: provider.RoleUser, Content: "hi"}}}) - got := lastBlockCacheControl(r) - if got == nil || got.TTL != "" { - t.Fatalf("cache_control = %+v, want default ttl for a 10s gap", got) + if string(got) != string(want) { + t.Fatalf("request bytes changed with unrelated context:\nfirst: %s\nsecond: %s", want, got) + } + if strings.Contains(string(got), `"ttl"`) { + t.Fatalf("default cache_control unexpectedly opted into a TTL: %s", got) } } func TestCacheControlOmitsTTLByDefault(t *testing.T) { - b, err := json.Marshal(ephemeral("")) + b, err := json.Marshal(ephemeral()) if err != nil { t.Fatalf("marshal: %v", err) } if string(b) != `{"type":"ephemeral"}` { t.Fatalf("default cache_control = %s, want byte-identical to every prior release", b) } - b, err = json.Marshal(ephemeral("1h")) - if err != nil { - t.Fatalf("marshal: %v", err) - } - if string(b) != `{"type":"ephemeral","ttl":"1h"}` { - t.Fatalf("1h cache_control = %s", b) - } -} - -func TestUseLongCacheTTL(t *testing.T) { - for _, tc := range []struct { - name string - gap time.Duration - hint bool - want bool - }{ - {"no hint", 0, false, false}, - {"below threshold", cacheGapLongTTLThreshold - time.Second, true, false}, - {"at threshold", cacheGapLongTTLThreshold, true, true}, - {"above threshold", cacheGapLongTTLThreshold + time.Minute, true, true}, - } { - ctx := context.Background() - if tc.hint { - ctx = provider.WithCacheGapHint(ctx, tc.gap) - } - if got := useLongCacheTTL(ctx); got != tc.want { - t.Errorf("%s: useLongCacheTTL = %v, want %v", tc.name, got, tc.want) - } - } } func TestConfiguredMaxOutputTokensRespectsMandatoryAnthropicFallback(t *testing.T) { @@ -277,6 +234,28 @@ func TestNewDetectsOfficialDeepSeekEndpoint(t *testing.T) { } } +func TestNewScopesNativeCacheWritePricingToAnthropic(t *testing.T) { + for _, tc := range []struct { + name string + baseURL string + want bool + }{ + {name: "default", want: true}, + {name: "official v1", baseURL: "https://api.anthropic.com/v1", want: true}, + {name: "compatible gateway", baseURL: "https://proxy.example.com/anthropic", want: false}, + } { + t.Run(tc.name, func(t *testing.T) { + p, err := New(provider.Config{Name: "test", BaseURL: tc.baseURL, Model: "claude-sonnet-4-6"}) + if err != nil { + t.Fatalf("New: %v", err) + } + if got := p.(*client).nativeAnthropic; got != tc.want { + t.Fatalf("nativeAnthropic = %v, want %v", got, tc.want) + } + }) + } +} + func TestMapStopReason(t *testing.T) { cases := map[string]string{ "end_turn": "stop", @@ -506,11 +485,46 @@ data: {"type":"message_stop"} if usage.CacheHitTokens != 7 || usage.CacheMissTokens != 18 { t.Fatalf("usage cache = hit %d miss %d", usage.CacheHitTokens, usage.CacheMissTokens) } + if usage.CacheWriteTokens != 5 || usage.CacheWriteBilledTokens != 0 { + t.Fatalf("compatible-gateway cache write = raw %d billed %v, want 5/0", usage.CacheWriteTokens, usage.CacheWriteBilledTokens) + } if usage.FinishReason != "stop" { t.Fatalf("finish reason = %q", usage.FinishReason) } } +func TestReadStreamPricesNativeCacheWritesAtDefaultTTL(t *testing.T) { + sse := `event: message_start +data: {"type":"message_start","message":{"usage":{"input_tokens":3,"cache_creation_input_tokens":5,"cache_read_input_tokens":7,"output_tokens":0}}} + +event: message_delta +data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}} + +event: message_stop +data: {"type":"message_stop"} +` + c := &client{name: "anthropic", nativeAnthropic: true} + resp := &http.Response{Body: io.NopCloser(strings.NewReader(sse))} + ch := make(chan provider.Chunk) + go c.readStream(context.Background(), resp, ch) + + var usage *provider.Usage + for ck := range ch { + if ck.Type == provider.ChunkError { + t.Fatalf("unexpected error chunk: %v", ck.Err) + } + if ck.Type == provider.ChunkUsage { + usage = ck.Usage + } + } + if usage == nil { + t.Fatal("expected a usage chunk") + } + if usage.CacheWriteTokens != 5 || usage.CacheWriteBilledTokens != 6.25 { + t.Fatalf("usage cache write = raw %d billed %v, want 5/6.25", usage.CacheWriteTokens, usage.CacheWriteBilledTokens) + } +} + // TestReadStreamError surfaces a mid-stream error event as a ChunkError. func TestReadStreamError(t *testing.T) { sse := "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"overloaded\"}}\n\n" diff --git a/internal/provider/cache_hint.go b/internal/provider/cache_hint.go deleted file mode 100644 index 66a0b0962c..0000000000 --- a/internal/provider/cache_hint.go +++ /dev/null @@ -1,34 +0,0 @@ -package provider - -import ( - "context" - "time" -) - -type cacheGapHintKey struct{} - -// WithCacheGapHint attaches how long it has been since this Agent last sent a -// provider request, so a provider with an opt-in longer cache TTL (Anthropic's -// 1h ephemeral breakpoint, versus its cheaper default 5m one) can decide -// whether the default window would likely already have expired. Mirrors the -// WithRetryNotify/WithRequestAttemptCounter context-key idiom in retry.go: a -// side channel for per-request signals that must never enter the request body -// or the cacheable prefix itself. -func WithCacheGapHint(ctx context.Context, gap time.Duration) context.Context { - if ctx == nil { - ctx = context.Background() - } - return context.WithValue(ctx, cacheGapHintKey{}, gap) -} - -// CacheGapHintFromContext returns the gap since the previous request and -// whether one was set. No value — the first request of an Agent's lifetime, a -// fresh sub-agent session, or a caller that never wired the hint — means -// default to the provider's cheaper, shorter cache TTL. -func CacheGapHintFromContext(ctx context.Context) (time.Duration, bool) { - if ctx == nil { - return 0, false - } - gap, ok := ctx.Value(cacheGapHintKey{}).(time.Duration) - return gap, ok -} diff --git a/internal/provider/openai/openai.go b/internal/provider/openai/openai.go index 08a31549ad..c863b517ac 100644 --- a/internal/provider/openai/openai.go +++ b/internal/provider/openai/openai.go @@ -630,6 +630,8 @@ func mergeUsage(total, next *provider.Usage, countRequests bool) *provider.Usage total.TotalTokens += next.TotalTokens total.CacheHitTokens += next.CacheHitTokens total.CacheMissTokens += next.CacheMissTokens + total.CacheWriteTokens += next.CacheWriteTokens + total.CacheWriteBilledTokens += next.CacheWriteBilledTokens total.ReasoningTokens += next.ReasoningTokens if countRequests { total.RequestCount = totalRequests + nextRequests diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index 68a6a116c8..6381048afa 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -76,12 +76,15 @@ func TestStreamRetriesThenSucceeds(t *testing.T) { } func TestMergeUsageCountsStreamsNotUsageChunks(t *testing.T) { - firstChunk := &provider.Usage{PromptTokens: 2, TotalTokens: 2, RequestCount: 2} - secondChunk := &provider.Usage{CompletionTokens: 1, TotalTokens: 1, RequestCount: 2} + firstChunk := &provider.Usage{PromptTokens: 2, TotalTokens: 2, RequestCount: 2, CacheWriteTokens: 2, CacheWriteBilledTokens: 2.5} + secondChunk := &provider.Usage{CompletionTokens: 1, TotalTokens: 1, RequestCount: 2, CacheWriteTokens: 3, CacheWriteBilledTokens: 6} oneStream := mergeUsage(firstChunk, secondChunk, false) if oneStream.RequestCount != 2 { t.Fatalf("same-stream request count = %d, want 2", oneStream.RequestCount) } + if oneStream.CacheWriteTokens != 5 || oneStream.CacheWriteBilledTokens != 8.5 { + t.Fatalf("same-stream cache writes = raw %d billed %v, want 5/8.5", oneStream.CacheWriteTokens, oneStream.CacheWriteBilledTokens) + } nextStream := &provider.Usage{PromptTokens: 3, TotalTokens: 3, RequestCount: 1} combined := mergeUsage(oneStream, nextStream, true) if combined.RequestCount != 3 { diff --git a/internal/provider/provider.go b/internal/provider/provider.go index 4516604f08..a7d5a8bf26 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -687,14 +687,16 @@ const ( // Estimated marks counts reconstructed locally because the provider's terminal // usage record did not arrive; exact provider usage leaves it false. type Usage struct { - PromptTokens int - CompletionTokens int - TotalTokens int - CacheHitTokens int // prompt tokens served from cache - CacheMissTokens int // prompt tokens not cached - ReasoningTokens int // subset of CompletionTokens spent on chain-of-thought - FinishReason string // "stop", "tool_calls", "length", "content_filter", "repetition_truncation", … - Estimated bool + PromptTokens int + CompletionTokens int + TotalTokens int + CacheHitTokens int // prompt tokens served from cache + CacheMissTokens int // prompt tokens not cached, including CacheWriteTokens + CacheWriteTokens int // subset of CacheMissTokens used to create provider cache entries + CacheWriteBilledTokens float64 // cache-write charge expressed in ordinary input-token equivalents + ReasoningTokens int // subset of CompletionTokens spent on chain-of-thought + FinishReason string // "stop", "tool_calls", "length", "content_filter", "repetition_truncation", … + Estimated bool // RequestCount is the number of provider requests represented by this // aggregate. Zero means one request for backward compatibility. Recovery // paths that merge multiple attempts set the exact count. @@ -757,8 +759,30 @@ func (p *Pricing) Cost(u *Usage) float64 { } else if miss == 0 && hit > 0 && u.PromptTokens > hit { miss = u.PromptTokens - hit } + // CacheMissTokens intentionally remains the raw prompt-token denominator + // used by cache hit-rate displays, so cache writes are included there. For + // cost, split those writes back out and replace them with their provider- + // supplied input-token equivalent (for example Anthropic's 1.25x 5-minute + // writes or 2x 1-hour writes). Older providers leave both fields at zero and + // keep the legacy one-input-rate behavior. A write count without billed + // units also falls back to 1x for backward compatibility. + write := u.CacheWriteTokens + if write < 0 { + write = 0 + } + if write > miss { + write = miss + } + billedWrite := 0.0 + if write > 0 { + billedWrite = u.CacheWriteBilledTokens + if billedWrite <= 0 { + billedWrite = float64(write) + } + } + inputTokenUnits := float64(miss-write) + billedWrite return (float64(hit)*p.CacheHit + - float64(miss)*p.Input + + inputTokenUnits*p.Input + float64(u.CompletionTokens)*p.Output) / 1e6 } diff --git a/internal/provider/provider_test.go b/internal/provider/provider_test.go index ddf1cda7a8..e59ab91c5b 100644 --- a/internal/provider/provider_test.go +++ b/internal/provider/provider_test.go @@ -433,6 +433,34 @@ func TestPricingCostCalculation(t *testing.T) { } } +func TestPricingCostUsesCacheWriteBillingTier(t *testing.T) { + p := &Pricing{Input: 2.0} + u := &Usage{ + CacheMissTokens: 500_000, + CacheWriteTokens: 100_000, + CacheWriteBilledTokens: 200_000, // 1h write at 2x input + } + // 400K ordinary misses + 100K cache writes billed as 200K input units. + if got := p.Cost(u); got != 1.2 { + t.Errorf("Cost = %f, want 1.2", got) + } +} + +func TestPricingCostCacheWriteFieldsAreBackwardCompatible(t *testing.T) { + p := &Pricing{Input: 2.0} + + // Old usage records have neither cache-write field and retain the original + // one-input-rate calculation. + if got := p.Cost(&Usage{CacheMissTokens: 500_000}); got != 1.0 { + t.Errorf("legacy Cost = %f, want 1.0", got) + } + // A producer that reports raw write tokens without a billing tier also + // falls back to the ordinary input rate instead of making writes free. + if got := p.Cost(&Usage{CacheMissTokens: 500_000, CacheWriteTokens: 100_000}); got != 1.0 { + t.Errorf("unpriced write Cost = %f, want 1.0", got) + } +} + func TestPricingCostFallsBackToPromptTokensAsMiss(t *testing.T) { p := &Pricing{Input: 2.0, Output: 10.0} u := &Usage{PromptTokens: 500_000, CompletionTokens: 100_000}