diff --git a/.claude/skills/forge.md b/.claude/skills/forge.md index 602336f4..265f6cda 100644 --- a/.claude/skills/forge.md +++ b/.claude/skills/forge.md @@ -212,7 +212,10 @@ threaded into context, stamped on every audit event: - `X-Invocation-Caller` — upstream caller identifier **Response headers** (FWS-3): `X-Forge-Tokens-In`, `X-Forge-Tokens-Out`, -`X-Forge-Duration-Ms`, `X-Forge-Model`, `X-Forge-Provider`. +`X-Forge-Duration-Ms`, `X-Forge-Model`, `X-Forge-Provider`. `X-Forge-Tokens-In` +bills from the TRUE input — summed `total_input_tokens` incl. Anthropic cache +read/creation (#431), guarded to never fall below the uncached delta — so a +cache-heavy stage can't slip past an orchestrator cost ceiling-check. **Agent Card** carries `name`, `description`, `url`, `version`, `protocolVersion: "0.3.0"`, `defaultInputModes` / @@ -279,7 +282,12 @@ Credentials read from `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` / `AWS_SESSI Token usage and request IDs are captured per provider at the call site and folded into the `llm_call` audit event (FWS-3) and into the per-invocation `LLMUsageAccumulator` so the response headers + the -final `invocation_complete` event carry totals. +final `invocation_complete` event carry totals. Under Anthropic prompt +caching the provider's `input_tokens` is only the uncached delta, so the +parser also captures `cache_read_input_tokens` / `cache_creation_input_tokens` +and every layer carries a summed `total_input_tokens` (= delta + cache +read + creation) as the bill-from figure (#431). OpenAI is unaffected — +its `prompt_tokens` already folds in cached input. **Read**: `docs/core-concepts/runtime-engine.md`, `forge-core/llm/`. @@ -1163,7 +1171,7 @@ when OTel tracing is enabled (OTel v1 / Phase 4 / #105). Both use | `AuditToolExec` | `tool_exec` | Tool execution `phase: start` / `phase: end`; carries `tool`, `args_size`, `result_size`, `duration_ms` | | `AuditEgressAllowed` | `egress_allowed` | Outbound request allowed (with domain, mode, source) | | `AuditEgressBlocked` | `egress_blocked` | Outbound request blocked | -| `AuditLLMCall` | `llm_call` | LLM provider call complete; `model`, `provider`, `input_tokens`, `output_tokens`, `duration_ms`, `request_id` | +| `AuditLLMCall` | `llm_call` | LLM provider call complete; `model`, `provider`, `input_tokens`, `output_tokens`, `total_input_tokens` (always; = input + cache read + creation — bill from this), `cache_read_input_tokens` / `cache_creation_input_tokens` (Anthropic caching only, omitempty), `duration_ms`, `request_id`. Under caching `input_tokens` is only the uncached delta (#431); `tokens_unavailable` keys off total input so a cache-read-only turn isn't misflagged as free | | `AuditLLMCallCancelled` | `llm_call_cancelled` | Streaming call aborted mid-flight; partial usage counts | | `AuditGuardrail` | `guardrail_check` | Mask / block / warn decision. Fields: `gate` (`input` / `context` / `tool_call` / `output` / `stream` — from library `Result.Gate`), `decision` (`masked` / `warned` / `blocked`), `guardrail`, `category`, `violation_count`, optional `tool`. Opt-in `evidence` (redacted + truncated triggering text) via `FORGE_GUARDRAIL_CAPTURE_EVIDENCE=true`. | | `AuditScheduleFire` | `schedule_fire` | Cron task triggered | @@ -1183,7 +1191,7 @@ when OTel tracing is enabled (OTel v1 / Phase 4 / #105). Both use | `context_compressed` | `context_compressed` | Context compression shrank content; `seam` (`tool_output` / `request`), `tool`, `tokens_before` / `tokens_after` / `saved_tokens` + running totals (tokenizer estimates) | | `context_expanded` | `context_expanded` | Model retrieved offloaded content via `context_expand`; `hash`, `hit`, `bytes`, producing `tool`, mined `candidates` (≤5, for fleet-wide learning aggregation) + running totals | | `context_pattern_suggested` | `context_pattern_suggested` | Learning loop surfaced a keep_patterns candidate (3+ expansions); `pattern`, `expansions`, `tools` | -| `AuditInvocationComplete` | `invocation_complete` | A2A invocation closed; `duration_ms`, `input_tokens_total`, `output_tokens_total`, `llm_call_count`, `model`, `provider` (FWS-3); with compression enabled also `compression_saved_tokens_total` (realized wire savings, compounds per history resend), `compression_event_saved_tokens`, `compression_count`, `expansion_count` | +| `AuditInvocationComplete` | `invocation_complete` | A2A invocation closed; `duration_ms`, `input_tokens_total`, `output_tokens_total`, `total_input_tokens_total` (bill-from sum incl. Anthropic cache read/creation, #431), `llm_call_count`, `model`, `provider` (FWS-3); with caching also `cache_read_input_tokens_total` / `cache_creation_input_tokens_total`; with compression enabled also `compression_saved_tokens_total` (realized wire savings, compounds per history resend), `compression_event_saved_tokens`, `compression_count`, `expansion_count` | | `AuditInvocationCancelled` | `invocation_cancelled` | A2A invocation cancelled via `tasks/cancel`; classified `reason` + partial token totals (FWS-4) | | `AuditTaskAdmissionDenied` | `task_admission_denied` | Inbound `tasks/send` denied by the platform admission middleware (#201; opt-in via `FORGE_ADMISSION_URL` + `FORGE_PLATFORM_TOKEN`); `reason`, `scope`, `window`, `reset_at`, `cached`. Caller sees HTTP 402 Payment Required. | | `AuditPolicyLoaded` | `policy_loaded` | One per non-empty policy layer at startup; `layer`, `source`, per-list size counters (FWS-5/6) | diff --git a/docs/core-concepts/hooks.md b/docs/core-concepts/hooks.md index 076f48ac..2f3723f8 100644 --- a/docs/core-concepts/hooks.md +++ b/docs/core-concepts/hooks.md @@ -111,10 +111,15 @@ The runner registers `AfterLLMCall` hooks that emit structured audit events for |-------|-------------| | `provider` | LLM provider name | | `model` | Model identifier | -| `input_tokens` | Prompt token count | +| `input_tokens` | Prompt token count. Under Anthropic prompt caching this is the **uncached delta only** — see `total_input_tokens` | | `output_tokens` | Completion token count | +| `cache_read_input_tokens` | Anthropic prompt-cache hit — cached-prefix tokens read this call (omitted when zero / non-Anthropic) | +| `cache_creation_input_tokens` | Anthropic prompt-cache write — tokens spent seeding the cache (omitted when zero / non-Anthropic) | +| `total_input_tokens` | `input_tokens` + cache read + creation — the true input consumption; **always present** (bill from this) | | `organization_id` | OpenAI Organization ID (when set) | +For the full prompt-caching rationale and the `tokens_unavailable` interaction, see [Token usage](../security/audit-logging.md#token-usage-and-execution-duration). + These events are logged via `slog` at Info level and can be consumed by external log aggregators for cost tracking and compliance. ## Progress Tracking diff --git a/docs/security/audit-logging.md b/docs/security/audit-logging.md index d3951958..ffe8401b 100644 --- a/docs/security/audit-logging.md +++ b/docs/security/audit-logging.md @@ -20,7 +20,7 @@ All runtime security events are emitted as structured NDJSON to stderr with corr | `llm_call` | LLM API call completed (with `input_tokens`, `output_tokens`, `model`, `provider`, `duration_ms`, `request_id`, and `fields.url` — the actual endpoint the request hit, e.g. a Kong base URL + `/v1/messages`; recorded even when payload capture is off since the URL is header-authed metadata, not payload). Any `user:pass@` userinfo in the base URL is **stripped** from the recorded `fields.url` so an inline-credential base URL doesn't leak into the audit stream (#358). See [Token usage and duration](#token-usage-and-execution-duration). | | `llm_call_cancelled` | Streaming LLM call cancelled mid-flight; carries partial token counts captured up to cancellation. | | `llm_call_failed` | An LLM API call failed (transport error or non-2xx) on the request path (#361). Carries `provider` / `model` / `duration_ms` and `fields.error` (the failure reason) — `fields.error` is **always** secret-scrubbed and length-capped regardless of the payload-capture toggle (see [What gets scrubbed](#what-gets-scrubbed)). Lets operators alert on provider/gateway outages without enabling payload capture. | -| `invocation_complete` | A2A invocation finished (auth → dispatch → engine → response). Carries `duration_ms` (wall-clock) plus aggregated `input_tokens_total` / `output_tokens_total` / `llm_call_count` / `model` / `provider`. When [context compression](../core-concepts/context-compression.md) is enabled it also carries `compression_saved_tokens_total` — REALIZED savings: tokens this invocation's LLM calls did not send because compression markers rode in place of originals, compounding on every resend of compressed history (this is the number that matches the provider bill) — plus `compression_event_saved_tokens` (the one-time per-compression deltas, matching the sum of this invocation's `context_compressed` events), `compression_count`, and `expansion_count` when nonzero. Accumulated per invocation by correlation ID so concurrent tasks never cross-contaminate. | +| `invocation_complete` | A2A invocation finished (auth → dispatch → engine → response). Carries `duration_ms` (wall-clock) plus aggregated `input_tokens_total` / `output_tokens_total` / `total_input_tokens_total` (the bill-from sum incl. Anthropic cache read/creation — see [Token usage](#token-usage-and-execution-duration)) / `llm_call_count` / `model` / `provider`. When prompt caching was active it also carries `cache_read_input_tokens_total` / `cache_creation_input_tokens_total`. When [context compression](../core-concepts/context-compression.md) is enabled it also carries `compression_saved_tokens_total` — REALIZED savings: tokens this invocation's LLM calls did not send because compression markers rode in place of originals, compounding on every resend of compressed history (this is the number that matches the provider bill) — plus `compression_event_saved_tokens` (the one-time per-compression deltas, matching the sum of this invocation's `context_compressed` events), `compression_count`, and `expansion_count` when nonzero. Accumulated per invocation by correlation ID so concurrent tasks never cross-contaminate. | | `invocation_cancelled` | A2A invocation cancelled mid-flight via `tasks/cancel` (or internal cancellation like parent ctx deadline). Carries `fields.reason` (one of `workflow_failure` / `cost_limit_exceeded` / `timeout` / `external_signal`), `duration_ms` up to cancellation, and any partial token totals consumed before the signal. See [Cancellation](#cancellation). | | `task_admission_denied` | A new inbound `tasks/send` was rejected by the platform admission middleware (issue #201; opt-in via `FORGE_ADMISSION_URL` + `FORGE_PLATFORM_TOKEN`). Carries `fields.reason` (platform-defined: `cost_limit_exceeded`, `billing_overdue`, …), `fields.scope` (`agent` / `workspace` / `org`), `fields.window` (`hourly` / `daily` / `monthly` / `billing_cycle`), `fields.reset_at` (RFC 3339), and `fields.cached` (`true` when served from the 5s per-agent cache). Caller observes HTTP 402 Payment Required with `Retry-After`. Since admission sits between auth and dispatch and emits via `EmitFromContext`, it carries the ingress-minted `correlation_id` (#278) — so admission denials group with the `auth_verify` of the same request in per-invocation views. See [Platform Admission Hook](admission.md). | | `guardrail_check` | Guardrail mask / block / warn decision. Carries `fields.gate` (`input` / `context` / `tool_call` / `output` / `stream` — sourced from the library `Result.Gate`), `fields.decision` (`masked` / `warned` / `blocked`), `fields.guardrail` + `fields.category` from the triggering violation, and `fields.violation_count`. `fields.tool` is present on `tool_call` and on `output` events for tool return text. With `FORGE_GUARDRAIL_CAPTURE_EVIDENCE=true` operators also opt into `fields.evidence` carrying the redacted + truncated triggering text. **Platform command denial (#238):** when a call matches a platform-policy `denied_command_patterns` entry, this event fires with `fields.source: "platform"`, `fields.guardrail: "platform_command_deny"`, `fields.pattern`, `fields.layer` (first-denying layer), `fields.policy_source` (file path), and the operator `fields.message` — the operator-authored, org-wide command control from [Platform Policy — Runtime command denial](platform-policy.md#runtime-command-denial). See [Guardrails — Audit Events](guardrails.md#audit-events). | @@ -132,6 +132,8 @@ See [Tenancy stamping reference](tenancy.md) for the precedence rules and the ag Every `llm_call` audit event carries the normalized token counts the provider returned in its response metadata, plus the wall-clock time spent in the provider call. Field naming aligns with [OTel GenAI semantic conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/) (`gen_ai.usage.input_tokens` / `gen_ai.usage.output_tokens`) so audit consumers can correlate Forge audit events with OTel traces without a translation table. +**Prompt caching and `total_input_tokens`.** When Anthropic prompt caching is active (Forge sets `cache_control` breakpoints on the tools + system prefix), the provider's `input_tokens` is only the **uncached delta** — the bulk of the prompt is billed separately as `cache_read_input_tokens` (a cache hit, ~10% rate) and `cache_creation_input_tokens` (the one-time write that seeds the cache). Reading `input_tokens` alone therefore undercounts real input by orders of magnitude on cache-heavy runs. To make correct usage the default, every `llm_call` also carries **`total_input_tokens` = `input_tokens` + `cache_read_input_tokens` + `cache_creation_input_tokens`** — always present, even when caching is off (where it equals `input_tokens`). Bill from `total_input_tokens`. OpenAI is unaffected: its `prompt_tokens` already folds cached input into the total, so the cache fields stay absent. + ```json { "ts": "2026-06-04T15:21:09Z", @@ -142,6 +144,9 @@ Every `llm_call` audit event carries the normalized token counts the provider re "provider": "anthropic", "input_tokens": 1240, "output_tokens": 387, + "cache_read_input_tokens": 18004, + "cache_creation_input_tokens": 512, + "total_input_tokens": 19756, "duration_ms": 2150, "request_id": "msg_01H8…" } @@ -149,9 +154,12 @@ Every `llm_call` audit event carries the normalized token counts the provider re | Field | Source | Notes | |---|---|---| -| `input_tokens` | Provider response usage | Maps to `gen_ai.usage.input_tokens` | +| `input_tokens` | Provider response usage | Maps to `gen_ai.usage.input_tokens`. Under prompt caching this is the **uncached delta only** — bill from `total_input_tokens` instead | | `output_tokens` | Provider response usage | Maps to `gen_ai.usage.output_tokens` | -| `tokens_unavailable` | Audit emitter | `true` when both counts are zero — some self-hosted Ollama setups don't return usage; billing consumers must distinguish "not measured" from "zero tokens used" | +| `cache_read_input_tokens` | Provider response usage | Anthropic prompt-cache hit — cached-prefix tokens read this call. Omitted when zero / non-Anthropic | +| `cache_creation_input_tokens` | Provider response usage | Anthropic prompt-cache write — tokens spent seeding the cache. Omitted when zero / non-Anthropic | +| `total_input_tokens` | Audit emitter | `input_tokens` + `cache_read_input_tokens` + `cache_creation_input_tokens` — the true input consumption. **Always present** (equals `input_tokens` when caching is off). This is the bill-from field | +| `tokens_unavailable` | Audit emitter | `true` when `total_input_tokens` **and** `output_tokens` are both zero — some self-hosted Ollama setups don't return usage; billing consumers must distinguish "not measured" from "zero tokens used". A cache-read-only turn (`input_tokens` 0 but `cache_read_input_tokens` > 0) is **not** flagged — it consumed real input | | `model` | Runtime model config | The model identifier the executor was configured with | | `provider` | Runtime model config | One of `anthropic`, `openai`, `ollama`, `custom` | | `duration_ms` | Captured at call site | Wall-clock time spent in `client.Chat`, in milliseconds | @@ -165,7 +173,7 @@ A2A response headers carry the same per-invocation totals inline so an orchestra | Header | Value | |---|---| -| `X-Forge-Tokens-In` | Sum of `input_tokens` across all LLM calls in the invocation | +| `X-Forge-Tokens-In` | Sum of `total_input_tokens` across all LLM calls in the invocation — the true input incl. Anthropic cache read/creation (#431), so an orchestrator's cost ceiling-check can't be fooled by a cache-heavy stage. Never reports below the summed uncached `input_tokens` delta | | `X-Forge-Tokens-Out` | Sum of `output_tokens` across all LLM calls in the invocation | | `X-Forge-Duration-Ms` | Wall-clock invocation duration (auth → dispatch → engine → response) | | `X-Forge-Model` | Most-recently-used model | @@ -193,6 +201,7 @@ Cancellation latency is bounded by the time for the current LLM call or tool cal "state": "canceled", "input_tokens_total": 940, "output_tokens_total": 215, + "total_input_tokens_total": 18744, "llm_call_count": 2, "model": "claude-sonnet-4-6", "provider": "anthropic" @@ -520,6 +529,8 @@ Every emitted event carries: | `workflow_id` / `workflow_execution_id` / `stage_id` / `step_id` / `invocation_caller` | string | optional | Populated when the request carried `X-Workflow-*` headers (FWS-2). `workflow_id` is the workflow definition (stable across runs); `workflow_execution_id` is the per-run instance (FORGE-2 / #185 split). | | `model` / `provider` | string | optional | LLM call attribution (FWS-3) | | `input_tokens` / `output_tokens` / `tokens_unavailable` | int / bool | optional | LLM call usage (FWS-3) | +| `total_input_tokens` | int | optional | True input = `input_tokens` + cache read + cache creation; the bill-from field. Present on every LLM call (#431) | +| `cache_read_input_tokens` / `cache_creation_input_tokens` | int | Anthropic caching only | Prompt-cache read (hit) / creation (write) token counts; omitted when zero (#431) | | `duration_ms` | int64 | optional | Wall-clock duration (FWS-3) | | `request_id` | string | optional | Provider-specific call identifier (FWS-3) | | `trace_id` / `span_id` | string | tracing-on only | W3C-format lowercase hex (32/16 chars) of the OTel span active at emit time. Pivots audit row ↔ trace tree. See [trace cross-link](#trace-cross-link-otel-v1-105). | diff --git a/forge-cli/runtime/forge_usage_headers.go b/forge-cli/runtime/forge_usage_headers.go index 7f1d2e77..2e4db004 100644 --- a/forge-cli/runtime/forge_usage_headers.go +++ b/forge-cli/runtime/forge_usage_headers.go @@ -32,7 +32,19 @@ func applyForgeUsageHeaders(h http.Header, snap coreruntime.LLMUsageSnapshot) { h.Set(HeaderForgeDurationMs, strconv.FormatInt(snap.InvocationDuration.Milliseconds(), 10)) return } - h.Set(HeaderForgeTokensIn, strconv.Itoa(snap.InputTokens)) + // Bill-from the TRUE input: uncached delta + Anthropic cache read + + // cache creation (issue #431). Under prompt caching snap.InputTokens + // is only the uncached delta, so an orchestrator ceiling-checking + // against X-Forge-Tokens-In would wildly under-count and let a + // cache-heavy stage sail past a cost cap. The max() guard keeps the + // header from ever reporting BELOW the uncached delta if a caller + // hands us a snapshot with TotalInputTokens unpopulated (mirrors + // security-next#36's `total_input_tokens ?? input_tokens` fallback). + tokensIn := snap.TotalInputTokens + if snap.InputTokens > tokensIn { + tokensIn = snap.InputTokens + } + h.Set(HeaderForgeTokensIn, strconv.Itoa(tokensIn)) h.Set(HeaderForgeTokensOut, strconv.Itoa(snap.OutputTokens)) h.Set(HeaderForgeDurationMs, strconv.FormatInt(snap.InvocationDuration.Milliseconds(), 10)) if snap.PrimaryModel != "" { diff --git a/forge-cli/runtime/forge_usage_headers_test.go b/forge-cli/runtime/forge_usage_headers_test.go index c7682755..c8f428f6 100644 --- a/forge-cli/runtime/forge_usage_headers_test.go +++ b/forge-cli/runtime/forge_usage_headers_test.go @@ -41,6 +41,42 @@ func TestApplyForgeUsageHeaders_StampsAllFields(t *testing.T) { } } +func TestApplyForgeUsageHeaders_TokensIn_BillsFromTrueTotalUnderCaching(t *testing.T) { + // Issue #431: under Anthropic prompt caching, InputTokens is only the + // uncached delta. X-Forge-Tokens-In must carry TotalInputTokens (delta + // + cache read + cache creation) so an orchestrator ceiling-checking + // against the header can't be fooled into letting a cache-heavy stage + // past a cost cap. + h := http.Header{} + applyForgeUsageHeaders(h, coreruntime.LLMUsageSnapshot{ + InputTokens: 32, // summed uncached delta + TotalInputTokens: 8032, // delta + 8000 cached prefix + OutputTokens: 180, + LLMCallCount: 2, + }) + if h.Get(HeaderForgeTokensIn) != "8032" { + t.Errorf("X-Forge-Tokens-In = %q, want 8032 (true total incl. cache), not the uncached delta", h.Get(HeaderForgeTokensIn)) + } + if h.Get(HeaderForgeTokensOut) != "180" { + t.Errorf("X-Forge-Tokens-Out = %q, want 180", h.Get(HeaderForgeTokensOut)) + } +} + +func TestApplyForgeUsageHeaders_TokensIn_NeverBelowUncachedDelta(t *testing.T) { + // Defensive: a snapshot with TotalInputTokens unpopulated (0) but a + // real InputTokens must still report the delta, never a smaller + // number — mirrors the `total_input_tokens ?? input_tokens` fallback. + h := http.Header{} + applyForgeUsageHeaders(h, coreruntime.LLMUsageSnapshot{ + InputTokens: 450, + OutputTokens: 180, + LLMCallCount: 1, + }) + if h.Get(HeaderForgeTokensIn) != "450" { + t.Errorf("X-Forge-Tokens-In = %q, want 450 (falls back to InputTokens when total is unset)", h.Get(HeaderForgeTokensIn)) + } +} + func TestApplyForgeUsageHeaders_NoLLMCalls_StillStampsDuration(t *testing.T) { // Short-circuited invocation (guardrail-failed before LLM dispatch): // orchestrator still wants a wall-clock figure, but token fields diff --git a/forge-cli/runtime/runner.go b/forge-cli/runtime/runner.go index 71e6eb4c..9dce2b68 100644 --- a/forge-cli/runtime/runner.go +++ b/forge-cli/runtime/runner.go @@ -1733,6 +1733,15 @@ func (r *Runner) registerHandlers(srv *server.Server, executor coreruntime.Agent if snap.LLMCallCount > 0 { fields["input_tokens_total"] = snap.InputTokens fields["output_tokens_total"] = snap.OutputTokens + // True input incl. Anthropic cache read/creation (issue #431); + // the cache breakdown is added only when caching was active. + fields["total_input_tokens_total"] = snap.TotalInputTokens + if snap.CacheReadInputTokens > 0 { + fields["cache_read_input_tokens_total"] = snap.CacheReadInputTokens + } + if snap.CacheCreationInputTokens > 0 { + fields["cache_creation_input_tokens_total"] = snap.CacheCreationInputTokens + } fields["llm_call_count"] = snap.LLMCallCount if snap.PrimaryModel != "" { fields["model"] = snap.PrimaryModel @@ -2018,6 +2027,15 @@ func (r *Runner) executeTask( if snap.LLMCallCount > 0 { fields["input_tokens_total"] = snap.InputTokens fields["output_tokens_total"] = snap.OutputTokens + // True input incl. Anthropic cache read/creation (issue #431); + // the cache breakdown is added only when caching was active. + fields["total_input_tokens_total"] = snap.TotalInputTokens + if snap.CacheReadInputTokens > 0 { + fields["cache_read_input_tokens_total"] = snap.CacheReadInputTokens + } + if snap.CacheCreationInputTokens > 0 { + fields["cache_creation_input_tokens_total"] = snap.CacheCreationInputTokens + } fields["llm_call_count"] = snap.LLMCallCount if snap.PrimaryModel != "" { fields["model"] = snap.PrimaryModel @@ -2307,6 +2325,15 @@ func (r *Runner) registerRESTHandlers(srv *server.Server, executor coreruntime.A if snap.LLMCallCount > 0 { fields["input_tokens_total"] = snap.InputTokens fields["output_tokens_total"] = snap.OutputTokens + // True input incl. Anthropic cache read/creation (issue #431); + // the cache breakdown is added only when caching was active. + fields["total_input_tokens_total"] = snap.TotalInputTokens + if snap.CacheReadInputTokens > 0 { + fields["cache_read_input_tokens_total"] = snap.CacheReadInputTokens + } + if snap.CacheCreationInputTokens > 0 { + fields["cache_creation_input_tokens_total"] = snap.CacheCreationInputTokens + } fields["llm_call_count"] = snap.LLMCallCount if snap.PrimaryModel != "" { fields["model"] = snap.PrimaryModel @@ -2707,6 +2734,11 @@ func (r *Runner) registerAuditHooks(hooks *coreruntime.HookRegistry, auditLogger usage.InputTokens = hctx.Response.Usage.InputTokens usage.OutputTokens = hctx.Response.Usage.OutputTokens usage.TotalTokens = hctx.Response.Usage.TotalTokens + // Carry Anthropic prompt-cache counts through so the llm_call + // event emits cache_read/creation + a summed total_input_tokens + // instead of undercounting to the uncached delta (issue #431). + usage.CacheReadInputTokens = hctx.Response.Usage.CacheReadInputTokens + usage.CacheCreationInputTokens = hctx.Response.Usage.CacheCreationInputTokens requestID = hctx.Response.ID } // FWS-8 payload-capture surfaces. Fields stays nil in the diff --git a/forge-core/llm/providers/anthropic.go b/forge-core/llm/providers/anthropic.go index d4fa2a09..b5fc693f 100644 --- a/forge-core/llm/providers/anthropic.go +++ b/forge-core/llm/providers/anthropic.go @@ -339,6 +339,14 @@ type anthropicResponse struct { Usage struct { InputTokens int `json:"input_tokens"` OutputTokens int `json:"output_tokens"` + // Prompt-cache counts. Under the cache_control breakpoints set in + // buildAnthropicRequest, input_tokens is only the uncached delta; + // the cached prefix is billed here — cache_read on a hit, and + // cache_creation on the write that seeds it. Dropping these + // undercounts real input by orders of magnitude on cache-heavy + // runs (issue #431). + CacheReadInputTokens int `json:"cache_read_input_tokens"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens"` } `json:"usage"` } @@ -379,9 +387,16 @@ func (c *AnthropicClient) parseAnthropicResponse(body io.Reader) (*llm.ChatRespo Model: resp.Model, Message: msg, Usage: llm.UsageInfo{ - InputTokens: resp.Usage.InputTokens, - OutputTokens: resp.Usage.OutputTokens, - TotalTokens: resp.Usage.InputTokens + resp.Usage.OutputTokens, + InputTokens: resp.Usage.InputTokens, + OutputTokens: resp.Usage.OutputTokens, + CacheReadInputTokens: resp.Usage.CacheReadInputTokens, + CacheCreationInputTokens: resp.Usage.CacheCreationInputTokens, + // True total: uncached delta + cache read + cache creation + + // output. Under caching the old input+output sum undercounted + // (issue #431); this mirrors OpenAI, whose prompt_tokens + // already folds cached input into the total. + TotalTokens: resp.Usage.InputTokens + resp.Usage.CacheReadInputTokens + + resp.Usage.CacheCreationInputTokens + resp.Usage.OutputTokens, }, FinishReason: finishReason, }, nil diff --git a/forge-core/llm/providers/usage_extraction_test.go b/forge-core/llm/providers/usage_extraction_test.go index c21f7f87..7f863e16 100644 --- a/forge-core/llm/providers/usage_extraction_test.go +++ b/forge-core/llm/providers/usage_extraction_test.go @@ -48,6 +48,55 @@ func TestAnthropic_PopulatesUsageWithOTelAlignedNames(t *testing.T) { } } +func TestAnthropic_PopulatesCacheTokens_TotalInputCountsCachedPrefix(t *testing.T) { + // Under prompt caching the Anthropic usage.input_tokens is only the + // uncached delta; the cached prefix is billed as cache_read / + // cache_creation. Dropping those undercounts real input by orders of + // magnitude (issue #431). The provider must surface them and fold + // them into TotalTokens. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + _ = json.NewEncoder(w).Encode(map[string]any{ + "id": "msg_cache", + "content": []map[string]any{{"type": "text", "text": "ok"}}, + "stop_reason": "end_turn", + "usage": map[string]int{ + "input_tokens": 12, + "output_tokens": 8, + "cache_read_input_tokens": 4000, + "cache_creation_input_tokens": 200, + }, + }) + })) + defer srv.Close() + + c := NewAnthropicClient(llm.ClientConfig{APIKey: "x", BaseURL: srv.URL, Model: "claude-3-5-sonnet"}) + resp, err := c.Chat(context.Background(), &llm.ChatRequest{ + Model: "claude-3-5-sonnet", + Messages: []llm.ChatMessage{{Role: llm.RoleUser, Content: "hi"}}, + }) + if err != nil { + t.Fatalf("Chat: %v", err) + } + if resp.Usage.InputTokens != 12 { + t.Errorf("InputTokens (uncached delta) = %d, want 12", resp.Usage.InputTokens) + } + if resp.Usage.CacheReadInputTokens != 4000 { + t.Errorf("CacheReadInputTokens = %d, want 4000", resp.Usage.CacheReadInputTokens) + } + if resp.Usage.CacheCreationInputTokens != 200 { + t.Errorf("CacheCreationInputTokens = %d, want 200", resp.Usage.CacheCreationInputTokens) + } + if got := resp.Usage.TotalInputTokens(); got != 4212 { + t.Errorf("TotalInputTokens() = %d, want 4212 (12+4000+200)", got) + } + // TotalTokens now folds the cached prefix + output, matching OpenAI's + // inclusive semantics: 12 + 4000 + 200 + 8. + if resp.Usage.TotalTokens != 4220 { + t.Errorf("TotalTokens = %d, want 4220 (input+cache_read+cache_creation+output)", resp.Usage.TotalTokens) + } +} + func TestOpenAI_PopulatesUsageWithOTelAlignedNames(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = io.Copy(io.Discard, r.Body) diff --git a/forge-core/llm/types.go b/forge-core/llm/types.go index 04fb5182..212258c0 100644 --- a/forge-core/llm/types.go +++ b/forge-core/llm/types.go @@ -96,4 +96,23 @@ type UsageInfo struct { InputTokens int `json:"input_tokens"` OutputTokens int `json:"output_tokens"` TotalTokens int `json:"total_tokens"` + + // Prompt-cache token counts (Anthropic). When prompt caching is + // active, the provider's input_tokens is only the UNCACHED delta; + // the cached prefix is billed separately as cache_read_input_tokens + // (hit, ~10% rate) and cache_creation_input_tokens (write, one-time). + // Recording them here lets the audit/usage layer report true input + // consumption instead of the delta alone (issue #431). Zero for + // providers that fold cached input into InputTokens (OpenAI's + // prompt_tokens already includes it) or when caching is off. + CacheReadInputTokens int `json:"cache_read_input_tokens,omitempty"` + CacheCreationInputTokens int `json:"cache_creation_input_tokens,omitempty"` +} + +// TotalInputTokens returns the true input consumption: the uncached +// delta plus cache-read plus cache-creation tokens. For providers whose +// InputTokens already includes cached input (OpenAI), the cache fields +// are zero and this equals InputTokens. +func (u UsageInfo) TotalInputTokens() int { + return u.InputTokens + u.CacheReadInputTokens + u.CacheCreationInputTokens } diff --git a/forge-core/runtime/audit.go b/forge-core/runtime/audit.go index 1eaaff15..29ff65c3 100644 --- a/forge-core/runtime/audit.go +++ b/forge-core/runtime/audit.go @@ -455,6 +455,18 @@ type AuditEvent struct { OutputTokens *int `json:"output_tokens,omitempty"` TokensUnavailable bool `json:"tokens_unavailable,omitempty"` + // Prompt-cache token breakdown + summed true input (issue #431). + // When Anthropic prompt caching is active, input_tokens is only the + // uncached delta; cache_read_input_tokens / cache_creation_input_tokens + // carry the cached prefix. total_input_tokens = input + cache_read + + // cache_creation is emitted whenever input_tokens is, so a consumer + // reading it alone (security-next#36's `total_input_tokens ?? input_tokens` + // fallback) can't undercount. The two cache fields use omitempty so + // non-cached calls and non-Anthropic providers keep the pre-#431 shape. + CacheReadInputTokens *int `json:"cache_read_input_tokens,omitempty"` + CacheCreationInputTokens *int `json:"cache_creation_input_tokens,omitempty"` + TotalInputTokens *int `json:"total_input_tokens,omitempty"` + // DurationMs is the wall-clock duration in milliseconds. Populated on // llm_call, tool_exec, and invocation_complete events. DurationMs *int64 `json:"duration_ms,omitempty"` @@ -1010,6 +1022,21 @@ type LLMUsage struct { InputTokens int OutputTokens int TotalTokens int + // Prompt-cache counts (Anthropic). InputTokens is only the uncached + // delta when caching is active; these carry the cached prefix so the + // emitted total_input_tokens reflects true input consumption instead + // of undercounting (issue #431). Zero for providers that fold cached + // input into InputTokens. + CacheReadInputTokens int + CacheCreationInputTokens int +} + +// TotalInputTokens returns the true input consumption for the call: +// uncached delta + cache read + cache creation. Downstream cost/usage +// consumers read this so a cache-heavy call is not undercounted by +// reading InputTokens (the delta) alone. +func (u LLMUsage) TotalInputTokens() int { + return u.InputTokens + u.CacheReadInputTokens + u.CacheCreationInputTokens } // EmitLLMCall builds and emits an llm_call (or llm_call_cancelled) @@ -1041,7 +1068,25 @@ func (a *AuditLogger) EmitLLMCall(ctx context.Context, args LLMCallAuditArgs) { in, out := args.Usage.InputTokens, args.Usage.OutputTokens evt.InputTokens = &in evt.OutputTokens = &out - if in == 0 && out == 0 { + // Emit total_input_tokens alongside input_tokens ALWAYS (even when it + // equals input_tokens, i.e. no caching): security-next#36 reads + // `total_input_tokens ?? input_tokens`, so a consistently-present field + // keeps that fallback on the fast path and prevents any reader from + // undercounting a cache-heavy call by reading the uncached delta alone + // (issue #431). + totalIn := args.Usage.TotalInputTokens() + evt.TotalInputTokens = &totalIn + // The cache breakdown is Anthropic-only detail — omitempty keeps the + // pre-#431 JSON shape for non-cached / non-Anthropic calls. + if cr := args.Usage.CacheReadInputTokens; cr != 0 { + evt.CacheReadInputTokens = &cr + } + if cc := args.Usage.CacheCreationInputTokens; cc != 0 { + evt.CacheCreationInputTokens = &cc + } + // Unavailable only when the provider reported NO usage at all — a + // cache-read-only turn (in==0 but cache_read>0) did consume input. + if totalIn == 0 && out == 0 { evt.TokensUnavailable = true } d := args.Duration.Milliseconds() diff --git a/forge-core/runtime/audit_llm_test.go b/forge-core/runtime/audit_llm_test.go index ce0bd445..eaeedf75 100644 --- a/forge-core/runtime/audit_llm_test.go +++ b/forge-core/runtime/audit_llm_test.go @@ -57,6 +57,107 @@ func TestEmitLLMCall_FullUsage(t *testing.T) { } } +func TestEmitLLMCall_CacheTokens_EmitsBreakdownAndSummedTotalInput(t *testing.T) { + // Issue #431: under Anthropic prompt caching, input_tokens is only + // the uncached delta. The llm_call event must carry the cache + // breakdown AND a summed total_input_tokens so a consumer reading + // total_input_tokens alone can't undercount. + var buf bytes.Buffer + audit := NewAuditLogger(&buf) + + audit.EmitLLMCall(context.Background(), LLMCallAuditArgs{ + Model: "claude-sonnet-4-6", + Provider: "anthropic", + Usage: LLMUsage{ + InputTokens: 12, + OutputTokens: 8, + CacheReadInputTokens: 4000, + CacheCreationInputTokens: 200, + }, + Duration: 10 * time.Millisecond, + }) + + var evt AuditEvent + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &evt); err != nil { + t.Fatalf("decode: %v\n%s", err, buf.String()) + } + if evt.InputTokens == nil || *evt.InputTokens != 12 { + t.Errorf("InputTokens (uncached delta) want 12, got %v", evt.InputTokens) + } + if evt.CacheReadInputTokens == nil || *evt.CacheReadInputTokens != 4000 { + t.Errorf("CacheReadInputTokens want 4000, got %v", evt.CacheReadInputTokens) + } + if evt.CacheCreationInputTokens == nil || *evt.CacheCreationInputTokens != 200 { + t.Errorf("CacheCreationInputTokens want 200, got %v", evt.CacheCreationInputTokens) + } + if evt.TotalInputTokens == nil || *evt.TotalInputTokens != 4212 { + t.Errorf("TotalInputTokens want 4212 (12+4000+200), got %v", evt.TotalInputTokens) + } + if evt.TokensUnavailable { + t.Errorf("TokensUnavailable must be false — the call consumed cached input") + } + // Wire-name check: the emitted JSON must use the exact field names + // security-next#36 reads. + js := buf.String() + for _, want := range []string{`"cache_read_input_tokens":4000`, `"cache_creation_input_tokens":200`, `"total_input_tokens":4212`} { + if !strings.Contains(js, want) { + t.Errorf("expected %s in JSON, got: %s", want, js) + } + } +} + +func TestEmitLLMCall_NoCaching_TotalInputEqualsInputAndCacheFieldsOmitted(t *testing.T) { + // Non-cached call (or non-Anthropic provider): total_input_tokens is + // still emitted (so the security-next fallback stays on its fast + // path) and equals input_tokens, while the two cache fields omit + // cleanly to preserve the pre-#431 JSON shape. + var buf bytes.Buffer + audit := NewAuditLogger(&buf) + audit.EmitLLMCall(context.Background(), LLMCallAuditArgs{ + Model: "gpt-4o", + Provider: "openai", + Usage: LLMUsage{InputTokens: 30, OutputTokens: 5, TotalTokens: 35}, + Duration: 1 * time.Millisecond, + }) + var evt AuditEvent + _ = json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &evt) + if evt.TotalInputTokens == nil || *evt.TotalInputTokens != 30 { + t.Errorf("TotalInputTokens want 30 (== input when no caching), got %v", evt.TotalInputTokens) + } + if evt.CacheReadInputTokens != nil || evt.CacheCreationInputTokens != nil { + t.Errorf("cache fields must omit when zero, got read=%v creation=%v", + evt.CacheReadInputTokens, evt.CacheCreationInputTokens) + } + js := buf.String() + for _, forbidden := range []string{`"cache_read_input_tokens"`, `"cache_creation_input_tokens"`} { + if strings.Contains(js, forbidden) { + t.Errorf("zero cache field %s must omit, got: %s", forbidden, js) + } + } +} + +func TestEmitLLMCall_CacheReadOnly_NotFlaggedUnavailable(t *testing.T) { + // A fully-cached turn reports input_tokens=0 but cache_read>0 — real + // input was consumed, so tokens_unavailable must stay false (else + // billing mistakes a large cached call for a free one). + var buf bytes.Buffer + audit := NewAuditLogger(&buf) + audit.EmitLLMCall(context.Background(), LLMCallAuditArgs{ + Model: "claude-sonnet-4-6", + Provider: "anthropic", + Usage: LLMUsage{InputTokens: 0, OutputTokens: 40, CacheReadInputTokens: 5000}, + Duration: 5 * time.Millisecond, + }) + var evt AuditEvent + _ = json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &evt) + if evt.TokensUnavailable { + t.Errorf("cache-read-only call consumed input; TokensUnavailable must be false, got %+v", evt) + } + if evt.TotalInputTokens == nil || *evt.TotalInputTokens != 5000 { + t.Errorf("TotalInputTokens want 5000, got %v", evt.TotalInputTokens) + } +} + func TestEmitLLMCall_TokensUnavailable_OllamaMissingUsage(t *testing.T) { // Self-hosted setups (some Ollama models) don't return token counts. // EmitLLMCall must flag tokens_unavailable=true rather than emit diff --git a/forge-core/runtime/usage_accumulator.go b/forge-core/runtime/usage_accumulator.go index c246b78d..3970eae8 100644 --- a/forge-core/runtime/usage_accumulator.go +++ b/forge-core/runtime/usage_accumulator.go @@ -20,15 +20,18 @@ import ( // whether OTel tracing is enabled — they're the orchestration channel, // not the observability channel. See issue #87 / FWS-3. type LLMUsageAccumulator struct { - mu sync.Mutex - invocationStart time.Time - inputTokensSum int - outputTokensSum int - llmTimeSum time.Duration - primaryModel string - primaryProvider string - llmCallCount int - tokensUnavailHit bool + mu sync.Mutex + invocationStart time.Time + inputTokensSum int + outputTokensSum int + cacheReadTokensSum int + cacheCreationTokensSum int + totalInputTokensSum int + llmTimeSum time.Duration + primaryModel string + primaryProvider string + llmCallCount int + tokensUnavailHit bool } // NewLLMUsageAccumulator returns a fresh accumulator with its invocation @@ -47,6 +50,12 @@ func (a *LLMUsageAccumulator) AddLLMCall(model, provider string, usage LLMUsage, defer a.mu.Unlock() a.inputTokensSum += usage.InputTokens a.outputTokensSum += usage.OutputTokens + a.cacheReadTokensSum += usage.CacheReadInputTokens + a.cacheCreationTokensSum += usage.CacheCreationInputTokens + // Aggregate TRUE input (delta + cache read + cache creation) so the + // run total reflects real consumption on cache-heavy runs, not the + // uncached delta (issue #431). + a.totalInputTokensSum += usage.TotalInputTokens() a.llmTimeSum += duration a.llmCallCount++ if model != "" { @@ -55,7 +64,7 @@ func (a *LLMUsageAccumulator) AddLLMCall(model, provider string, usage LLMUsage, if provider != "" { a.primaryProvider = provider } - if usage.InputTokens == 0 && usage.OutputTokens == 0 { + if usage.TotalInputTokens() == 0 && usage.OutputTokens == 0 { a.tokensUnavailHit = true } } @@ -64,14 +73,21 @@ func (a *LLMUsageAccumulator) AddLLMCall(model, provider string, usage LLMUsage, // at a single point in time. Returned by Snapshot for use by the A2A // response handler. type LLMUsageSnapshot struct { - InputTokens int - OutputTokens int - LLMTimeTotal time.Duration // sum of per-LLM-call durations - InvocationDuration time.Duration // wall-clock since accumulator creation - PrimaryModel string - PrimaryProvider string - LLMCallCount int - TokensUnavailable bool + InputTokens int + OutputTokens int + // Cache breakdown + true input aggregate (issue #431). InputTokens is + // the summed uncached delta; TotalInputTokens = InputTokens + + // CacheReadInputTokens + CacheCreationInputTokens is the real input + // consumption for the invocation. All zero for non-caching providers. + CacheReadInputTokens int + CacheCreationInputTokens int + TotalInputTokens int + LLMTimeTotal time.Duration // sum of per-LLM-call durations + InvocationDuration time.Duration // wall-clock since accumulator creation + PrimaryModel string + PrimaryProvider string + LLMCallCount int + TokensUnavailable bool } // Snapshot returns the current totals. Safe to call from a goroutine @@ -80,14 +96,17 @@ func (a *LLMUsageAccumulator) Snapshot() LLMUsageSnapshot { a.mu.Lock() defer a.mu.Unlock() return LLMUsageSnapshot{ - InputTokens: a.inputTokensSum, - OutputTokens: a.outputTokensSum, - LLMTimeTotal: a.llmTimeSum, - InvocationDuration: time.Since(a.invocationStart), - PrimaryModel: a.primaryModel, - PrimaryProvider: a.primaryProvider, - LLMCallCount: a.llmCallCount, - TokensUnavailable: a.tokensUnavailHit && a.inputTokensSum == 0 && a.outputTokensSum == 0, + InputTokens: a.inputTokensSum, + OutputTokens: a.outputTokensSum, + CacheReadInputTokens: a.cacheReadTokensSum, + CacheCreationInputTokens: a.cacheCreationTokensSum, + TotalInputTokens: a.totalInputTokensSum, + LLMTimeTotal: a.llmTimeSum, + InvocationDuration: time.Since(a.invocationStart), + PrimaryModel: a.primaryModel, + PrimaryProvider: a.primaryProvider, + LLMCallCount: a.llmCallCount, + TokensUnavailable: a.tokensUnavailHit && a.totalInputTokensSum == 0 && a.outputTokensSum == 0, } } diff --git a/forge-core/runtime/usage_accumulator_test.go b/forge-core/runtime/usage_accumulator_test.go index 35d166b4..4e89da4f 100644 --- a/forge-core/runtime/usage_accumulator_test.go +++ b/forge-core/runtime/usage_accumulator_test.go @@ -30,6 +30,48 @@ func TestLLMUsageAccumulator_AggregatesAcrossCalls(t *testing.T) { } } +func TestLLMUsageAccumulator_SumsCacheTokensAndTrueInput(t *testing.T) { + // Issue #431: the run aggregate must sum total_input_tokens (delta + + // cache read + cache creation), not just the uncached delta, so a + // cache-heavy run's invocation_complete reflects real consumption. + acc := NewLLMUsageAccumulator() + // Cold call: seeds the cache (creation), small delta. + acc.AddLLMCall("claude", "anthropic", + LLMUsage{InputTokens: 12, OutputTokens: 8, CacheCreationInputTokens: 4000}, time.Millisecond) + // Warm call: reads the cached prefix, small fresh delta. + acc.AddLLMCall("claude", "anthropic", + LLMUsage{InputTokens: 20, OutputTokens: 10, CacheReadInputTokens: 4000}, time.Millisecond) + + snap := acc.Snapshot() + if snap.InputTokens != 32 { + t.Errorf("InputTokens (uncached delta sum) = %d, want 32", snap.InputTokens) + } + if snap.CacheCreationInputTokens != 4000 { + t.Errorf("CacheCreationInputTokens sum = %d, want 4000", snap.CacheCreationInputTokens) + } + if snap.CacheReadInputTokens != 4000 { + t.Errorf("CacheReadInputTokens sum = %d, want 4000", snap.CacheReadInputTokens) + } + if snap.TotalInputTokens != 8032 { + t.Errorf("TotalInputTokens = %d, want 8032 (32 delta + 4000 creation + 4000 read)", snap.TotalInputTokens) + } +} + +func TestLLMUsageAccumulator_CacheReadOnlyDoesNotLatchUnavailable(t *testing.T) { + // A run made entirely of fully-cached turns (input delta 0, cache + // read > 0) consumed real input — TokensUnavailable must stay false. + acc := NewLLMUsageAccumulator() + acc.AddLLMCall("claude", "anthropic", + LLMUsage{InputTokens: 0, OutputTokens: 40, CacheReadInputTokens: 5000}, time.Millisecond) + snap := acc.Snapshot() + if snap.TokensUnavailable { + t.Errorf("cache-read-only run consumed input; TokensUnavailable must be false, got %+v", snap) + } + if snap.TotalInputTokens != 5000 { + t.Errorf("TotalInputTokens = %d, want 5000", snap.TotalInputTokens) + } +} + func TestLLMUsageAccumulator_PrimaryIsMostRecentNonEmpty(t *testing.T) { // Spec: X-Forge-Model / X-Forge-Provider report "the primary model // used (most recent if multiple)." This matches the most common