Anthropic streaming path drops input + cache tokens (StreamDelta.Usage carries only output_tokens)

[initializ-mk]

Problem

The Anthropic streaming path (readAnthropicStream, forge-core/llm/providers/anthropic.go) never captures input tokens. It handles only the message_delta event, which carries output_tokens:

case "message_delta":
    var ev anthropicMessageDelta   // Usage struct is { OutputTokens int }
    ...
    ch <- llm.StreamDelta{
        FinishReason: finishReason,
        Usage:        &llm.UsageInfo{ OutputTokens: ev.Usage.OutputTokens },
    }

There is no message_start case — and message_start is exactly where Anthropic reports the input usage for a streaming call: message.usage.input_tokens, cache_read_input_tokens, and cache_creation_input_tokens. So on the streaming path:

• input_tokens is dropped (not just the cached prefix — the whole input),
• cache_read_input_tokens / cache_creation_input_tokens are dropped,
• StreamDelta.Usage only ever carries OutputTokens.

This is the streaming sibling of #431 (which fixed the non-streaming parseAnthropicResponse path), plus a pre-existing gap that input tokens were never captured on streaming at all.

Blast radius (why this is latent, not live-billing)

The A2A agent runtime does not stream: the executor's ExecuteStream (forge-core/runtime/loop.go) just wraps Execute, which calls e.client.Chat (non-streaming → parseAnthropicResponse, already fixed in #431). The AfterLLMCall hook that feeds the llm_call audit event + LLMUsageAccumulator + X-Forge-Tokens-In reads that non-streaming response.

The only production caller of the provider ChatStream is forge-cli/cmd/ui.go (the workspace UI chat), which does not feed the audit/billing accumulator. So no live billing path undercounts today.

The risk is regression-latency: the moment any A2A/billing path switches to ChatStream (or a streaming client is wired into the accumulator), input + cache tokens silently read as zero. A cache-heavy streaming invocation would bill as ~free, and tokens_unavailable would even latch true.

Fix

1. Add a message_start case to readAnthropicStream that parses message.usage (input_tokens, cache_read_input_tokens, cache_creation_input_tokens) and emits a StreamDelta carrying them on UsageInfo (reuse the TotalInputTokens() helper + inclusive TotalTokens semantics added in Anthropic llm_call undercounts tokens: capture cache_read/creation + emit total_input_tokens (parity with initializ-sdk#10) #431).
2. Merge the message_start input usage with the message_delta output usage so the final accumulated UsageInfo for the stream is complete (Anthropic splits input at start, output at delta).
3. Whichever layer accumulates StreamDelta.Usage into the final ChatResponse.Usage must sum input from message_start and output from message_delta — verify a streaming ChatResponse.Usage matches the equivalent non-streaming call.

Acceptance

• A streaming Anthropic call's final UsageInfo carries input_tokens + cache_read_input_tokens + cache_creation_input_tokens + output_tokens, matching the non-streaming path for the same prompt.
• TotalInputTokens() on a cache-heavy streaming call equals the true input, not zero.
• OpenAI streaming unaffected (already emits usage on the final chunk).

Related

• Anthropic llm_call undercounts tokens: capture cache_read/creation + emit total_input_tokens (parity with initializ-sdk#10) #431 — the non-streaming fix (merged/in-flight). This issue closes the parallel streaming gap.
• initializ-sdk#10 — the SDK-side equivalent.

Flagged as a non-blocking follow-up in the #431 (PR #432) review.