feat(usage): cache_read_tokens and cache_write_tokens on usage events

CHANGELOG.md

@@ -9,6 +9,10 @@ All notable changes to light-tokenmeter are listed here. The format follows
 ## [Unreleased]
 
 ### Added
+- `cache_read_tokens` and `cache_write_tokens` on usage events (`usage.v1` fields 17 and 18) and
+  summed in summary and by-meta rows: the input tokens read from and written to the vendor's
+  prompt cache, which vendors bill apart from plain input. They are part of `tokens_input`.
+  Stored in Postgres and ClickHouse; events from older callers carry 0.
 - Authentication: bearer tokens with `write` and `read` scopes, configured as SHA-256 hashes in
   `TOKENMETER_TOKENS` or `TOKENMETER_TOKENS_FILE`, several at once for rotation. Refusals are
   `401`/`403` over HTTP and `UNAUTHENTICATED`/`PERMISSION_DENIED` over gRPC. Without tokens the

README.md

@@ -142,6 +142,7 @@ An event describes one model call:
 | `vendor_http_status` | int | no | HTTP status returned by the vendor. |
 | `meta` | map of string | no | Your own tags, e.g. `{"job_id": "…"}`: non-empty keys, at most 16 KiB of keys and values per event. |
 | `cost_nanousd` | int ≥0 | no | Call cost in nano-USD (1e-9 USD), priced by the caller; stored as-is and never recomputed. |
+| `cache_read_tokens`, `cache_write_tokens` | int ≥0 | no | Input tokens read from and written to the vendor's prompt cache. They are part of `tokens_input`, not added to it, so together they must not exceed it; an event that breaks this is refused. Anthropic reports them outside its `input_tokens`: send the sum of all three as `tokens_input`. |
 
 Batches (`POST /v1/usage/events:batch`, or `RecordUsageEvents` over gRPC) take up to 500
 events and report how many were created and how many were duplicates:
@@ -183,7 +184,8 @@ curl -s "localhost:8080/v1/usage/summary?$Q&model=gpt-4o&group_by_user_id=true"
 ```
 
 Each row carries `tokens_input`, `tokens_output`, `tokens_total`, `request_count`,
-`sum_duration_ms` and `cost_nanousd` (summed spend; divide by 1e9 for USD), so you can
+`sum_duration_ms`, `cost_nanousd` (summed spend; divide by 1e9 for USD) and
+`cache_read_tokens` / `cache_write_tokens` (the cached part of `tokens_input`), so you can
 compute tokens per request, cost per request, average latency and error share yourself. The signals that usually separate greedy from worthy:
 
 - high `tokens_total / request_count` on a trivial `action_id` (bloated prompts),

internal/database/migrations/008_cache_tokens.sql

@@ -0,0 +1,15 @@
+-- +goose Up
+-- +goose StatementBegin
+-- Parts of tokens_input, kept apart because vendors bill cached input at a
+-- fraction of the input rate (and cache writes above it).
+ALTER TABLE usage_events
+    ADD COLUMN IF NOT EXISTS cache_read_tokens BIGINT NOT NULL DEFAULT 0,
+    ADD COLUMN IF NOT EXISTS cache_write_tokens BIGINT NOT NULL DEFAULT 0;
+COMMENT ON COLUMN usage_events.cache_read_tokens IS 'Input tokens read from the prompt cache (part of tokens_input)';
+COMMENT ON COLUMN usage_events.cache_write_tokens IS 'Input tokens written to the prompt cache (part of tokens_input)';
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+ALTER TABLE usage_events DROP COLUMN IF EXISTS cache_read_tokens, DROP COLUMN IF EXISTS cache_write_tokens;
+-- +goose StatementEnd

internal/domain/event.go

@@ -69,6 +69,10 @@ type UsageEvent struct {
 	VendorHTTPStatus int32
 	Meta             map[string]string
 	CostNanoUSD      int64
+	// Prompt-cache tokens, both part of TokensInput: read from the vendor's
+	// cache, and written to it, each priced differently from plain input.
+	CacheReadTokens  int64
+	CacheWriteTokens int64
 }
 
 // Validation errors
@@ -86,6 +90,7 @@ var (
 	ErrVendorTooLong     = errors.New("vendor exceeds 64 characters")
 	ErrModelTooLong      = errors.New("model exceeds 128 characters")
 	ErrTokensMismatch    = errors.New("tokens_total must equal tokens_input + tokens_output")
+	ErrCacheTokens       = errors.New("cache_read_tokens and cache_write_tokens must be >= 0 and together not exceed tokens_input")
 	ErrNegativeCost      = errors.New("cost_nanousd must not be negative")
 	ErrOccurredAtRange   = errors.New("occurred_at must be between 1900-01-01 and 2262-04-10")
 	ErrErrorClassTooLong = errors.New("error_class exceeds 256 characters")
@@ -150,6 +155,12 @@ func (e *UsageEvent) Validate() error {
 	if e.CostNanoUSD < 0 {
 		return ErrNegativeCost
 	}
+	// Checked only when sent, so events without them validate as before; the
+	// subtraction cannot overflow once 0 <= read <= input.
+	if (e.CacheReadTokens != 0 || e.CacheWriteTokens != 0) && (e.CacheReadTokens < 0 || e.CacheWriteTokens < 0 ||
+		e.CacheReadTokens > e.TokensInput || e.CacheWriteTokens > e.TokensInput-e.CacheReadTokens) {
+		return ErrCacheTokens
+	}
 	if len(e.ErrorClass) > MaxStringLength {
 		return ErrErrorClassTooLong
 	}

internal/domain/meta_usage.go

@@ -25,4 +25,7 @@ type MetaUsageRow struct {
 	TokensTotal  int64
 	RequestCount int64
 	CostNanoUSD  int64
+	// Parts of TokensInput served from and written to the prompt cache.
+	CacheReadTokens  int64
+	CacheWriteTokens int64
 }

internal/domain/summary.go

@@ -43,4 +43,7 @@ type SummaryRow struct {
 	RequestCount  int64
 	SumDurationMs int64
 	CostNanoUSD   int64
+	// Parts of TokensInput served from and written to the prompt cache.
+	CacheReadTokens  int64
+	CacheWriteTokens int64
 }

internal/repository/outbox.go

@@ -179,7 +179,8 @@ func (r *Repository) OutboxBacklog(ctx context.Context) (int64, time.Duration, e
 
 const eventColumns = `event_id, created_at, occurred_at, user_id, project_id, action_id,
 	vendor, model, status, duration_ms, tokens_input, tokens_output, tokens_total,
-	COALESCE(error_class, ''), COALESCE(vendor_http_status, 0), meta, cost_nanousd`
+	COALESCE(error_class, ''), COALESCE(vendor_http_status, 0), meta, cost_nanousd,
+	cache_read_tokens, cache_write_tokens`
 
 func (r *Repository) eventsByID(ctx context.Context, ids []uuid.UUID) ([]*domain.UsageEvent, error) {
 	rows, err := r.pool.Query(ctx, `SELECT `+eventColumns+` FROM usage_events WHERE event_id = ANY($1)`, ids)
@@ -200,7 +201,8 @@ func scanEvents(rows pgx.Rows) ([]*domain.UsageEvent, error) {
 		)
 		if err := rows.Scan(&e.EventID, &e.CreatedAt, &e.OccurredAt, &e.UserID, &e.ProjectID, &e.ActionID,
 			&e.Vendor, &e.Model, &status, &e.DurationMs, &e.TokensInput, &e.TokensOutput, &e.TokensTotal,
-			&e.ErrorClass, &e.VendorHTTPStatus, &meta, &e.CostNanoUSD); err != nil {
+			&e.ErrorClass, &e.VendorHTTPStatus, &meta, &e.CostNanoUSD,
+			&e.CacheReadTokens, &e.CacheWriteTokens); err != nil {
 			return nil, fmt.Errorf("scan outbox event: %w", err)
 		}
 		e.Status = domain.Status(status)

internal/repository/outbox_test.go

@@ -19,7 +19,7 @@ func outboxEvent(meta map[string]string) *domain.UsageEvent {
 		EventID: uuid.New(), OccurredAt: &now, UserID: "u", ProjectID: "p", ActionID: "a",
 		Vendor: "openai", Model: "m", Status: domain.StatusError, DurationMs: 12,
 		TokensInput: 3, TokensOutput: 4, TokensTotal: 7, ErrorClass: "rate_limited", VendorHTTPStatus: 429,
-		Meta: meta, CostNanoUSD: 99,
+		Meta: meta, CostNanoUSD: 99, CacheReadTokens: 2, CacheWriteTokens: 1,
 	}
 }
 
@@ -87,6 +87,8 @@ func TestOutboxClaimAckAndTakeOver(t *testing.T) {
 	require.Equal(t, first.ErrorClass, stored.ErrorClass)
 	require.Equal(t, first.VendorHTTPStatus, stored.VendorHTTPStatus)
 	require.Equal(t, first.CostNanoUSD, stored.CostNanoUSD)
+	require.Equal(t, first.CacheReadTokens, stored.CacheReadTokens)
+	require.Equal(t, first.CacheWriteTokens, stored.CacheWriteTokens)
 	require.False(t, stored.CreatedAt.IsZero())
 
 	require.NoError(t, repo.AckOutbox(ctx, b1.ID))

internal/repository/postgres_test.go

@@ -198,18 +198,20 @@ func TestCostAggregation(t *testing.T) {
 
 	newEvent := func(model string, cost int64) *domain.UsageEvent {
 		return &domain.UsageEvent{
-			EventID:      uuid.New(),
-			UserID:       "u1",
-			ProjectID:    "p1",
-			ActionID:     "a1",
-			Vendor:       "openai",
-			Model:        model,
-			Status:       domain.StatusSuccess,
-			TokensInput:  10,
-			TokensOutput: 20,
-			TokensTotal:  30,
-			Meta:         map[string]string{"job_id": "job-cost"},
-			CostNanoUSD:  cost,
+			EventID:          uuid.New(),
+			UserID:           "u1",
+			ProjectID:        "p1",
+			ActionID:         "a1",
+			Vendor:           "openai",
+			Model:            model,
+			Status:           domain.StatusSuccess,
+			TokensInput:      10,
+			TokensOutput:     20,
+			TokensTotal:      30,
+			Meta:             map[string]string{"job_id": "job-cost"},
+			CostNanoUSD:      cost,
+			CacheReadTokens:  6,
+			CacheWriteTokens: 2,
 		}
 	}
 
@@ -229,19 +231,28 @@ func TestCostAggregation(t *testing.T) {
 	})
 	require.NoError(t, err)
 	byModel := map[string]int64{}
+	var cacheRead, cacheWrite int64
 	for _, r := range summary {
 		byModel[r.Model] += r.CostNanoUSD
+		cacheRead += r.CacheReadTokens
+		cacheWrite += r.CacheWriteTokens
 	}
+	assert.Equal(t, int64(18), cacheRead, "InsertOne and InsertBatch both store cache tokens")
+	assert.Equal(t, int64(6), cacheWrite)
 	assert.Equal(t, int64(3_000_000), byModel["gpt-4o"])
 	assert.Equal(t, int64(150_000), byModel["gpt-4o-mini"])
 
 	metaRows, _, err := repo.QueryUsageByMeta(ctx, domain.MetaUsageParams{MetaKey: "job_id", MetaValue: "job-cost"})
 	require.NoError(t, err)
-	var total int64
+	var total, metaCacheRead, metaCacheWrite int64
 	for _, r := range metaRows {
 		total += r.CostNanoUSD
+		metaCacheRead += r.CacheReadTokens
+		metaCacheWrite += r.CacheWriteTokens
 	}
 	assert.Equal(t, int64(3_150_000), total)
+	assert.Equal(t, int64(18), metaCacheRead)
+	assert.Equal(t, int64(6), metaCacheWrite)
 }
 
 // A retried batch inserts nothing new, and InsertBatch says so by ID, so the