fix(usage): record UsageLog on the streaming path too

Stage A.1 hooked only Router.Chat; the production agent uses
Router.ChatStream, so after deploy usage_log stayed empty despite
"usage logging enabled" and successful LLM calls.

Fix plumbs usage through the streaming path end to end:

* StreamChunk gets a Usage field populated on the final Done chunk.

* openai_compat.readSSE parses the SSE terminal chunk's usage block
  (OpenRouter sends one with choices=[] when usage.include=true),
  including cached_tokens and reasoning_tokens from the nested
  *_details sub-objects. Bumps the SSE scanner buffer to 1MB so long
  reasoning traces don't hide the usage chunk.

* Router.ChatStream wraps the provider's channel with
  instrumentedStream: it tees chunks through to the caller, and when
  the Done chunk arrives it calls the same recordUsage routine
  Router.Chat uses. Start-of-stream errors and sync/fallback paths
  also get a UsageLog row (with error_class set) so dashboards
  surface failed calls.

After this, every LLM request — streamed or not, successful or not —
produces exactly one usage_log record per provider hop.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Alexey Panfilov

internal/llm/openai_compat.go

@@ -140,13 +140,15 @@ type rawChatRequest struct {
 
 // rawStreamDelta is an SSE chunk from an OpenAI-compatible streaming response.
 type rawStreamDelta struct {
+	ID      string `json:"id"`
 	Choices []struct {
 		Delta struct {
 			Content   string        `json:"content"`
 			ToolCalls []rawToolCall `json:"tool_calls"`
 		} `json:"delta"`
 		FinishReason *string `json:"finish_reason"`
 	} `json:"choices"`
+	Usage *rawUsage `json:"usage"` // last chunk (usually after finish_reason) carries this when usage.include=true
 }
 
 // rawMessage uses `any` for Content so we can serialize null, string, or array.
@@ -455,8 +457,14 @@ func (p *openAICompatProvider) readSSE(resp *http.Response, ch chan<- StreamChun
 
 	// Accumulate tool calls across deltas (they arrive in fragments).
 	var toolCallsByIdx = make(map[int]*rawToolCall)
+	var usage Usage
+	var haveUsage bool
 
 	scanner := bufio.NewScanner(resp.Body)
+	// Stream responses can include large payloads (reasoning traces, long tool
+	// arguments). Default Scanner buffer is 64KB; bump to 1MB so we don't miss
+	// the final usage chunk on long responses.
+	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
 	for scanner.Scan() {
 		line := scanner.Text()
 		if !strings.HasPrefix(line, "data: ") {
@@ -471,6 +479,24 @@ func (p *openAICompatProvider) readSSE(resp *http.Response, ch chan<- StreamChun
 		if err := json.Unmarshal([]byte(data), &delta); err != nil {
 			continue
 		}
+
+		// Usage block: OpenRouter sends a terminal chunk with choices=[] and
+		// a populated usage object when usage.include=true.
+		if delta.Usage != nil {
+			usage.PromptTokens = delta.Usage.PromptTokens
+			usage.CompletionTokens = delta.Usage.CompletionTokens
+			if delta.Usage.PromptTokensDetails != nil {
+				usage.CachedPromptTokens = delta.Usage.PromptTokensDetails.CachedTokens
+			}
+			if delta.Usage.CompletionTokensDetails != nil {
+				usage.ReasoningTokens = delta.Usage.CompletionTokensDetails.ReasoningTokens
+			}
+			haveUsage = true
+		}
+		if delta.ID != "" && usage.RequestID == "" {
+			usage.RequestID = delta.ID
+		}
+
 		if len(delta.Choices) == 0 {
 			continue
 		}
@@ -517,6 +543,9 @@ func (p *openAICompatProvider) readSSE(resp *http.Response, ch chan<- StreamChun
 			})
 		}
 	}
+	if haveUsage {
+		final.Usage = usage
+	}
 	if err := scanner.Err(); err != nil {
 		final.Err = err
 	}

internal/llm/provider.go

@@ -68,6 +68,7 @@ type StreamChunk struct {
 	ToolCalls []ToolCall // populated only in the final Done chunk if the LLM returned tool calls
 	Done      bool       // true when the stream is finished
 	Err       error      // non-nil if the stream encountered an error
+	Usage     Usage      // populated on the final Done chunk when the provider reports token counts
 }
 
 // StreamProvider is an optional interface for providers that support streaming.

internal/llm/router.go

@@ -436,30 +436,64 @@ func splitProviderName(name string) (provider, model string) {
 // Falls back to wrapping a synchronous Chat() in a single-chunk channel.
 func (r *Router) ChatStream(ctx context.Context, messages []Message, systemPrompt string, tools []Tool) (<-chan StreamChunk, error) {
 	provider := r.pick(ctx, messages)
+	primaryKey := r.keyFor(provider)
 	r.mu.Lock()
 	r.lastRouted = provider.Name()
-	r.lastRoutedKey = r.keyFor(provider)
+	r.lastRoutedKey = primaryKey
 	r.mu.Unlock()
 	if sp, ok := provider.(StreamProvider); ok {
 		ch, err := sp.ChatStream(ctx, messages, systemPrompt, tools)
-		if err != nil && isUnavailable(err) {
-			// Try fallback providers synchronously.
-			return r.syncFallbackStream(ctx, provider, messages, systemPrompt, tools, err)
+		if err != nil {
+			// Even a start-of-stream failure deserves a usage record with
+			// error_class set so dashboards see it.
+			r.recordUsage(ctx, provider, primaryKey, Response{}, err, 0)
+			if isUnavailable(err) {
+				return r.syncFallbackStream(ctx, provider, messages, systemPrompt, tools, err)
+			}
+			return nil, err
 		}
-		return ch, err
+		return r.instrumentedStream(ctx, provider, primaryKey, ch), nil
 	}
 	// Provider does not stream — wrap synchronous call.
 	return r.syncStream(ctx, provider, messages, systemPrompt, tools)
 }
 
-// syncStream wraps a synchronous Chat() call in a channel.
+// instrumentedStream tees the provider's stream to the caller and records a
+// UsageLog entry when the terminal chunk arrives. Usage data comes from the
+// provider (openai_compat parses it out of the SSE payload); when absent the
+// record still lands with zero tokens so dashboards show the call happened.
+func (r *Router) instrumentedStream(ctx context.Context, provider Provider, role string, upstream <-chan StreamChunk) <-chan StreamChunk {
+	out := make(chan StreamChunk, 64)
+	start := time.Now()
+	go func() {
+		defer close(out)
+		var finalResp Response
+		var finalErr error
+		for chunk := range upstream {
+			out <- chunk
+			if chunk.Done {
+				finalResp = Response{ToolCalls: chunk.ToolCalls, Usage: chunk.Usage}
+				finalErr = chunk.Err
+			}
+		}
+		r.recordUsage(ctx, provider, role, finalResp, finalErr, time.Since(start))
+	}()
+	return out
+}
+
+// syncStream wraps a synchronous Chat() call in a channel. Usage is recorded
+// via the same path as Router.Chat so streaming callers that hit non-stream
+// providers still produce UsageLog rows.
 func (r *Router) syncStream(ctx context.Context, provider Provider, messages []Message, systemPrompt string, tools []Tool) (<-chan StreamChunk, error) {
+	role := r.keyFor(provider)
+	start := time.Now()
 	resp, err := provider.Chat(ctx, messages, systemPrompt, tools)
+	r.recordUsage(ctx, provider, role, resp, err, time.Since(start))
 	if err != nil {
 		return nil, err
 	}
 	ch := make(chan StreamChunk, 1)
-	ch <- StreamChunk{Delta: resp.Content, ToolCalls: resp.ToolCalls, Done: true}
+	ch <- StreamChunk{Delta: resp.Content, ToolCalls: resp.ToolCalls, Done: true, Usage: resp.Usage}
 	close(ch)
 	return ch, nil
 }
@@ -479,12 +513,13 @@ func (r *Router) syncFallbackStream(ctx context.Context, failed Provider, messag
 		if r.OnFallback != nil {
 			r.OnFallback(failed.Name(), next.Name())
 		}
-		// Try streaming on fallback if supported.
+		// Try streaming on fallback if supported — still instrumented.
 		if sp, ok := next.(StreamProvider); ok {
 			ch, err := sp.ChatStream(ctx, messages, systemPrompt, tools)
 			if err == nil {
-				return ch, nil
+				return r.instrumentedStream(ctx, next, "fallback", ch), nil
 			}
+			r.recordUsage(ctx, next, "fallback", Response{}, err, 0)
 		}
 		// Otherwise sync.
 		return r.syncStream(ctx, next, messages, systemPrompt, tools)