fix(tracing): stamp prompt-cache token attributes on the LLM span (#441)

docs/core-concepts/observability-tracing.md

@@ -220,7 +220,9 @@ Forge mixes OTel GenAI semconv with Forge-specific `forge.*` namespaced attribut
 | `gen_ai.request.model` | `agent.execute`, `llm.completion` | requested model |
 | `gen_ai.response.model` | `llm.completion` | vendor-reported model (falls back to request model) |
 | `gen_ai.response.id` | `llm.completion` | provider completion id |
-| `gen_ai.usage.input_tokens` / `.output_tokens` | `llm.completion` | provider usage block |
+| `gen_ai.usage.input_tokens` / `.output_tokens` | `llm.completion` | provider usage block. Under Anthropic prompt caching `input_tokens` is only the uncached delta |
+| `gen_ai.usage.cache_read_input_tokens` / `.cache_creation_input_tokens` | `llm.completion` | Anthropic prompt-cache hit / write tokens. Stamped only when non-zero; absent for non-caching / non-Anthropic calls (#441) |
+| `gen_ai.usage.total_input_tokens` | `llm.completion` | `input + cache_read + cache_creation` — the bill-from sum, matching the `llm_call` audit event's `total_input_tokens`. Present only when caching contributed (#441) |
 | `gen_ai.response.finish_reasons` | `llm.completion` | provider stop reason |
 | `gen_ai.tool.name` | `tool.<tool_name>` | tool function name |
 | `gen_ai.tool.call.id` | `tool.<tool_name>` | LLM-assigned tool-call id |
@@ -240,7 +242,7 @@ Prompts, completions, tool args, and tool results are **off by default** — Pha
 |---|---|---|
 | (always) | `agent.execute` | `gen_ai.provider.name`, `gen_ai.agent.id`, `gen_ai.agent.name`, `gen_ai.agent.version`, `gen_ai.conversation.id`, `gen_ai.request.model` |
 | `capture_content: true` | `agent.execute` | `gen_ai.tool.definitions` (JSON array of the tool catalog available to the agent — potentially large, hence opt-in) |
-| (always) | `llm.completion` | `gen_ai.operation.name` (`chat`), `gen_ai.provider.name`, `gen_ai.request.model`, `gen_ai.response.model`, `gen_ai.response.id`, `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, `gen_ai.response.finish_reasons` |
+| (always) | `llm.completion` | `gen_ai.operation.name` (`chat`), `gen_ai.provider.name`, `gen_ai.request.model`, `gen_ai.response.model`, `gen_ai.response.id`, `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, `gen_ai.response.finish_reasons` (+ `gen_ai.usage.cache_read_input_tokens` / `.cache_creation_input_tokens` / `.total_input_tokens` when Anthropic prompt caching contributed, #441) |
 | `capture_content: true` | `llm.completion` | `gen_ai.input.messages` (JSON array of role+content pairs sent to the model), `gen_ai.output.messages` (JSON single-element array of role+content for the model's response) — current OTel GenAI semconv, supersedes the deprecated flat-string `gen_ai.prompt` / `gen_ai.completion` |
 | (always) | `tool.<name>` | `gen_ai.operation.name` (`execute_tool`), `gen_ai.tool.name`, `gen_ai.tool.call.id`, `gen_ai.tool.type`, `mcp.method.name` (MCP only), `error.type` (on failure) |
 | `capture_content: true` | `tool.<name>` | `gen_ai.tool.call.arguments` (raw arguments JSON), `gen_ai.tool.call.result` (raw output), `gen_ai.tool.description` (from the tool definition) |

docs/security/audit-logging.md

@@ -440,7 +440,7 @@ streams:
 
 | Pivot direction | How |
 |---|---|
-| **audit row → trace** | Paste the row's `trace_id` into Tempo / Jaeger / Honeycomb to land on the matching trace. Paste the `span_id` to jump directly to the span (an `llm_call` row's `span_id` resolves to the `llm.completion` span carrying matching `gen_ai.usage.*` tokens). |
+| **audit row → trace** | Paste the row's `trace_id` into Tempo / Jaeger / Honeycomb to land on the matching trace. Paste the `span_id` to jump directly to the span (an `llm_call` row's `span_id` resolves to the `llm.completion` span carrying matching `gen_ai.usage.*` tokens — input, output, and, under Anthropic prompt caching, `cache_read_input_tokens` / `cache_creation_input_tokens` / `total_input_tokens` too, #441). |
 | **trace → audit row** | Copy `trace_id` from a trace browser; grep the audit log for the corresponding row to get the FWS-8 payload metadata the trace does not carry. |
 
 Format: lowercase hex matching W3C `traceparent` semantics — 32-char

forge-core/observability/attrs.go

@@ -53,6 +53,21 @@ const (
 	AttrGenAIUsageInputTokens  = "gen_ai.usage.input_tokens"
 	AttrGenAIUsageOutputTokens = "gen_ai.usage.output_tokens"
 
+	// Prompt-cache usage (Anthropic; #441). Under caching input_tokens is
+	// only the uncached delta — these carry the cached prefix so traces
+	// show cache stats and stay consistent with the llm_call audit event
+	// (#431/#432). OTel GenAI semconv does not standardize cache-token
+	// attributes yet, so these follow forge's existing gen_ai.usage.*
+	// prefix. Stamped only when non-zero; absent for non-caching /
+	// non-Anthropic calls.
+	//   - CacheRead     : cached-prefix tokens read this call (a cache hit)
+	//   - CacheCreation : tokens spent seeding the cache (a cache write)
+	//   - TotalInput    : input + cache_read + cache_creation — the bill-from
+	//                     sum, matching the audit event's total_input_tokens
+	AttrGenAIUsageCacheReadInputTokens     = "gen_ai.usage.cache_read_input_tokens"
+	AttrGenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation_input_tokens"
+	AttrGenAIUsageTotalInputTokens         = "gen_ai.usage.total_input_tokens"
+
 	// AttrGenAIResponseFinishReasons mirrors the Anthropic/OpenAI
 	// "stop_reason" / "finish_reason" — "stop", "tool_use",
 	// "max_tokens", "end_turn", etc.

forge-core/runtime/loop.go

@@ -528,10 +528,25 @@ func (e *LLMExecutor) Execute(ctx context.Context, task *a2a.Task, msg *a2a.Mess
 		// then close the span. Doing this BEFORE the AfterLLMCall hook
 		// keeps the hook's redaction / audit work outside the LLM
 		// span's duration — the span is the provider call alone.
-		llmSpan.SetAttributes(
+		usageAttrs := []attribute.KeyValue{
 			attribute.Int(observability.AttrGenAIUsageInputTokens, resp.Usage.InputTokens),
 			attribute.Int(observability.AttrGenAIUsageOutputTokens, resp.Usage.OutputTokens),
-		)
+		}
+		// Prompt-cache stats (#441): stamp only when present so non-caching /
+		// non-Anthropic calls keep their span shape. Mirrors the llm_call
+		// audit event (#431/#432) so a span and its audit row agree on cache
+		// tokens. total_input_tokens is emitted whenever caching contributed,
+		// matching the audit event's always-present bill-from field.
+		if resp.Usage.CacheReadInputTokens > 0 {
+			usageAttrs = append(usageAttrs, attribute.Int(observability.AttrGenAIUsageCacheReadInputTokens, resp.Usage.CacheReadInputTokens))
+		}
+		if resp.Usage.CacheCreationInputTokens > 0 {
+			usageAttrs = append(usageAttrs, attribute.Int(observability.AttrGenAIUsageCacheCreationInputTokens, resp.Usage.CacheCreationInputTokens))
+		}
+		if resp.Usage.CacheReadInputTokens > 0 || resp.Usage.CacheCreationInputTokens > 0 {
+			usageAttrs = append(usageAttrs, attribute.Int(observability.AttrGenAIUsageTotalInputTokens, resp.Usage.TotalInputTokens()))
+		}
+		llmSpan.SetAttributes(usageAttrs...)
 		if resp.ID != "" {
 			llmSpan.SetAttributes(attribute.String(observability.AttrGenAIResponseID, resp.ID))
 		}

forge-core/runtime/loop_spans_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/initializ/forge/forge-core/a2a"
 	"github.com/initializ/forge/forge-core/llm"
 	"github.com/initializ/forge/forge-core/observability"
+	"go.opentelemetry.io/otel/attribute"
 )
 
 // TestExecuteEmitsHappyPathSpanTree pins the Phase 3 (#104) instrumentation
@@ -168,6 +169,105 @@ func TestExecuteEmitsHappyPathSpanTree(t *testing.T) {
 	}
 }
 
+// TestExecuteStampsCacheTokensOnLLMSpan pins #441: when the provider
+// reports prompt-cache tokens, the llm.completion span carries
+// gen_ai.usage.cache_read_input_tokens / cache_creation_input_tokens
+// and the summed total_input_tokens — so traces show cache stats and
+// agree with the llm_call audit event. A non-caching call must NOT emit
+// zero-valued cache attributes (span shape stays stable for OpenAI etc.).
+func TestExecuteStampsCacheTokensOnLLMSpan(t *testing.T) {
+	tp, rec := observability.NewTestTracerProvider()
+	SetTracerProvider(tp)
+	t.Cleanup(func() {
+		ResetTracerProviderForTest()
+		_ = tp.Shutdown(context.Background())
+	})
+
+	callCount := 0
+	client := &mockLLMClient{
+		chatFunc: func(_ context.Context, _ *llm.ChatRequest) (*llm.ChatResponse, error) {
+			callCount++
+			if callCount == 1 {
+				// Cache-heavy turn: small uncached delta + large cached prefix.
+				return &llm.ChatResponse{
+					Message: llm.ChatMessage{Role: llm.RoleAssistant, ToolCalls: []llm.ToolCall{{
+						ID: "tc-1", Type: "function", Function: llm.FunctionCall{Name: "echo", Arguments: `{"x":1}`},
+					}}},
+					Usage: llm.UsageInfo{
+						InputTokens: 12, OutputTokens: 8,
+						CacheReadInputTokens: 4000, CacheCreationInputTokens: 200,
+					},
+					FinishReason: "tool_calls",
+				}, nil
+			}
+			// Non-caching turn: no cache fields.
+			return &llm.ChatResponse{
+				Message:      llm.ChatMessage{Role: llm.RoleAssistant, Content: "done"},
+				Usage:        llm.UsageInfo{InputTokens: 30, OutputTokens: 5},
+				FinishReason: "stop",
+			}, nil
+		},
+	}
+	tools := &mockToolExecutor{executeFunc: func(_ context.Context, _ string, _ json.RawMessage) (string, error) {
+		return "echoed", nil
+	}}
+	exec := NewLLMExecutor(LLMExecutorConfig{
+		Client: client, Tools: tools, MaxIterations: 5, ModelName: "claude-test", Provider: "anthropic",
+	})
+
+	task := &a2a.Task{ID: "task-cache"}
+	msg := &a2a.Message{Role: a2a.MessageRoleUser, Parts: []a2a.Part{{Kind: a2a.PartKindText, Text: "hi"}}}
+	if _, err := exec.Execute(context.Background(), task, msg); err != nil {
+		t.Fatalf("Execute: %v", err)
+	}
+
+	llmSpans := rec.FindSpans("llm.completion")
+	if len(llmSpans) != 2 {
+		t.Fatalf("got %d llm.completion spans; want 2", len(llmSpans))
+	}
+
+	// attrsOf collapses a span's int attributes into a lookup keyed by
+	// attribute name, with a presence flag so we can distinguish "absent"
+	// from "present and zero".
+	attrsOf := func(s interface{ Attributes() []attribute.KeyValue }) (map[string]int64, map[string]bool) {
+		vals := map[string]int64{}
+		present := map[string]bool{}
+		for _, kv := range s.Attributes() {
+			vals[string(kv.Key)] = kv.Value.AsInt64()
+			present[string(kv.Key)] = true
+		}
+		return vals, present
+	}
+
+	// First span (cache-heavy) — carries the cache breakdown + summed total.
+	v, p := attrsOf(llmSpans[0])
+	if v[observability.AttrGenAIUsageInputTokens] != 12 {
+		t.Errorf("input_tokens = %d, want 12 (uncached delta)", v[observability.AttrGenAIUsageInputTokens])
+	}
+	if v[observability.AttrGenAIUsageCacheReadInputTokens] != 4000 {
+		t.Errorf("cache_read = %d, want 4000", v[observability.AttrGenAIUsageCacheReadInputTokens])
+	}
+	if v[observability.AttrGenAIUsageCacheCreationInputTokens] != 200 {
+		t.Errorf("cache_creation = %d, want 200", v[observability.AttrGenAIUsageCacheCreationInputTokens])
+	}
+	if v[observability.AttrGenAIUsageTotalInputTokens] != 4212 {
+		t.Errorf("total_input_tokens = %d, want 4212 (12+4000+200)", v[observability.AttrGenAIUsageTotalInputTokens])
+	}
+
+	// Second span (no caching) — cache attributes must be ABSENT, not zero.
+	_, p2 := attrsOf(llmSpans[1])
+	for _, k := range []string{
+		observability.AttrGenAIUsageCacheReadInputTokens,
+		observability.AttrGenAIUsageCacheCreationInputTokens,
+		observability.AttrGenAIUsageTotalInputTokens,
+	} {
+		if p2[k] {
+			t.Errorf("non-caching span must omit %q, but it was present", k)
+		}
+	}
+	_ = p // first-span presence not asserted individually beyond values above
+}
+
 // TestExecuteRecordsLLMErrorOnSpan confirms that when the provider's
 // Chat() returns an error, the llm.completion span records it (status
 // = Error, error event present) AND the outer agent.execute span's