Method mapStreamError

aibridge/intercept/messages/streaming.go:609–632

mapStreamError converts a mid-stream upstream error or processing error into a relayable ResponseError. Returns nil when the error is unrecoverable, in which case nothing can be relayed back.

(ctx context.Context, logger slog.Logger, streamErr, lastErr error)

// when the error is unrecoverable, in which case nothing can be
// relayed back.
func (*StreamingInterception) mapStreamError(ctx context.Context, logger slog.Logger, streamErr, lastErr error) *ResponseError {
	if streamErr != nil {
		if eventstream.IsUnrecoverableError(streamErr) {
			logger.Debug(ctx, "stream terminated", slog.Error(streamErr))
			// We can't reflect an error back if there's a connection error or the request context was canceled.
			return nil
		}
		if antErr := responseErrorFromAPIError(streamErr); antErr != nil {
			logger.Warn(ctx, "anthropic stream error", slog.Error(streamErr))
			return antErr
		}
		logger.Warn(ctx, "unknown stream error", slog.Error(streamErr))
		// Unfortunately, the Anthropic SDK does not support parsing errors received in the stream
		// into known types (i.e. [shared.OverloadedError]).
		// See https://github.com/anthropics/anthropic-sdk-go/blob/v1.12.0/packages/ssestream/ssestream.go#L172-L174
		// All it does is wrap the payload in an error - which is all we can return, currently.
		return newResponseError(fmt.Sprintf("unknown stream error: %s", streamErr), string(constant.ValueOf[constant.Error]()), http.StatusBadGateway, 0)
	}
	if lastErr != nil {
		logger.Warn(ctx, "stream processing failed", slog.Error(lastErr))
		return newResponseError(fmt.Sprintf("processing error: %s", lastErr), string(constant.ValueOf[constant.Error]()), http.StatusBadGateway, 0)
	}
	return nil
}

func (i *StreamingInterception) marshalEvent(event anthropic.MessageStreamEventUnion) ([]byte, error) {
	sj, err := sjson.Set(event.RawJSON(), "message.id", i.ID().String())