// Local helper: attach billed usage accumulated in-context to the error.
func attachBilledUsageFromContext(ctx *schemas.BifrostContext, bifrostErr *schemas.BifrostError) {
  usage, ok := ctx.Value(schemas.BifrostContextKeyStreamAccumulatedUsage).(*schemas.BifrostLLMUsage)
  if !ok || usage == nil { return }
  if usage.PromptTokens == 0 && usage.CompletionTokens == 0 && usage.TotalTokens == 0 { return }
  // Important: copy or snapshot if usage will be mutated later.
  u := *usage
  bifrostErr.ExtraFields.BilledUsage = &u
}

// Inside stream read loop:
var sawDone bool
for {
  data, err := sseReader.ReadDataLine()
  if err != nil {
    if errors.Is(err, io.EOF) {
      break
    }
    // pre-first-byte / upstream classification happens at the Do() boundary,
    // but for mid-stream read errors, enrich and send.
    e := providerUtils.NewBifrostOperationError("stream read failed", err)
    attachBilledUsageFromContext(ctx, e)
    providerUtils.ProcessAndSendBifrostError(ctx, postHookRunner, e, responseChan, logger, postHookSpanFinalizer)
    return
  }

  var evt CartesiaSSEEvent
  if err := sonic.Unmarshal(data, &evt); err != nil {
    logger.Warn("Failed to parse SSE event: %v", err)
    continue // log-and-continue
  }

  switch evt.Type {
  case "done":
    sawDone = true
    // optionally process done payload
  case "chunk":
    if evt.Data == "" { continue }
    audio, decErr := base64.StdEncoding.DecodeString(evt.Data)
    if decErr != nil {
      logger.Warn("Failed to decode chunk: %v", decErr)
      continue // keep stream alive on single chunk decode errors
    }
    // send delta chunk...
  case "error":
    // surface terminal provider error
    e := providerUtils.NewBifrostOperationError("cartesia stream error: "+evt.Error, nil)
    attachBilledUsageFromContext(ctx, e)
    providerUtils.ProcessAndSendBifrostError(ctx, postHookRunner, e, responseChan, logger, postHookSpanFinalizer)
    return
  }

  if sawDone { break }
}

// Terminal-invariant guard: EOF without terminal done is truncated.
if !sawDone {
  ctx.SetValue(schemas.BifrostContextKeyStreamEndIndicator, true)
  e := providerUtils.NewBifrostOperationError("stream ended before terminal done", io.ErrUnexpectedEOF)
  attachBilledUsageFromContext(ctx, e)
  providerUtils.ProcessAndSendBifrostError(ctx, postHookRunner, e, responseChan, logger, postHookSpanFinalizer)
  return
}