1010 lines
33 KiB
Go
1010 lines
33 KiB
Go
package gateway
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"gpt-load/internal/dialect"
|
|
"gpt-load/internal/execution"
|
|
platformheader "gpt-load/internal/platform/httpheader"
|
|
"gpt-load/internal/platform/redact"
|
|
"gpt-load/internal/protocol"
|
|
"gpt-load/internal/usage"
|
|
)
|
|
|
|
// ExecutionForwarder adapts the provider-neutral executor to the gateway's
|
|
// downstream commit boundary while the handler orchestration remains intact.
|
|
type ExecutionForwarder struct {
|
|
executor execution.Executor
|
|
representation *responseProcessor
|
|
usageCapture *usageCaptureBoundary
|
|
writeTimeout time.Duration
|
|
}
|
|
|
|
func NewExecutionForwarder(executor execution.Executor) *ExecutionForwarder {
|
|
return &ExecutionForwarder{
|
|
executor: executor, representation: &responseProcessor{redactor: redact.New()},
|
|
usageCapture: newUsageCaptureBoundary(),
|
|
writeTimeout: downstreamWriteTimeout,
|
|
}
|
|
}
|
|
|
|
func (forwarder *ExecutionForwarder) OpenWebsocket(ctx context.Context, input ForwardInput) (execution.WebsocketSession, execution.WebsocketResult) {
|
|
spec, err := newExecutionAttemptSpec(input)
|
|
if forwarder != nil && err == nil {
|
|
if opener, ok := forwarder.executor.(execution.WebsocketOpener); ok {
|
|
return opener.OpenWebsocket(ctx, spec)
|
|
}
|
|
}
|
|
return nil, execution.WebsocketResult{DispatchState: execution.DispatchNotSent, Error: &execution.ErrorEvidence{
|
|
Kind: execution.ErrorKindInvalidRequest, OriginHint: execution.ErrorOriginInternal,
|
|
ScopeHint: execution.ErrorScopeRequest, Code: "websocket_not_supported", Summary: "Native WebSocket is not supported.",
|
|
}}
|
|
}
|
|
|
|
func (forwarder *ExecutionForwarder) Forward(
|
|
ctx context.Context,
|
|
input ForwardInput,
|
|
) UpstreamResult {
|
|
spec, err := newExecutionAttemptSpec(input)
|
|
if err != nil || forwarder == nil || forwarder.executor == nil {
|
|
return executionInputFailure(err)
|
|
}
|
|
executionResult := forwarder.executor.Execute(ctx, spec)
|
|
if err := executionResult.Validate(); err != nil {
|
|
return invalidExecutionAttemptResult(executionResult)
|
|
}
|
|
result := upstreamFromExecutionResult(ctx, input, executionResult)
|
|
result = forwarder.prepareBufferedResult(input, result)
|
|
if (input.ClientProtocol == protocol.OpenAIImages ||
|
|
input.ClientProtocol == protocol.OpenAIEmbeddings || input.ClientProtocol == protocol.Rerank) && input.ObserveUsage &&
|
|
result.HasResponse() && result.StatusCode >= http.StatusOK &&
|
|
result.StatusCode < http.StatusMultipleChoices &&
|
|
executionResult.Usage == nil && result.Usage.State == usage.StateMissing && forwarder.usageCapture != nil {
|
|
result.Usage = forwarder.usageCapture.extractNonStreamingPlain(
|
|
input.Dialect,
|
|
result.ClassificationBody,
|
|
)
|
|
}
|
|
if input.ClientProtocol == protocol.OpenAIEmbeddings && result.Err == nil &&
|
|
result.StatusCode >= http.StatusOK && result.StatusCode < http.StatusMultipleChoices {
|
|
result.ClassificationBody = nil
|
|
}
|
|
return result
|
|
}
|
|
|
|
func invalidExecutionAttemptResult(result execution.AttemptResult) UpstreamResult {
|
|
dispatchState := execution.DispatchNotSent
|
|
if result.DispatchState == execution.DispatchLocal {
|
|
dispatchState = execution.DispatchLocal
|
|
} else if result.DispatchState == execution.DispatchMaybeSent ||
|
|
result.ResponseStarted || result.StatusCode != 0 || len(result.Header) > 0 || len(result.Body) > 0 {
|
|
dispatchState = execution.DispatchMaybeSent
|
|
}
|
|
evidence := execution.ErrorEvidence{
|
|
Kind: execution.ErrorKindInternal,
|
|
OriginHint: execution.ErrorOriginInternal,
|
|
ScopeHint: execution.ErrorScopeRequest,
|
|
Code: "attempt_result_contract_invalid",
|
|
Summary: "Attempt executor returned an invalid result.",
|
|
ReplaySafety: execution.ReplaySafetyUnknown,
|
|
}
|
|
return UpstreamResult{
|
|
Err: fmt.Errorf("%w: invalid attempt executor result", ErrUpstreamProtocol),
|
|
RequestWritten: dispatchState == execution.DispatchMaybeSent,
|
|
DispatchState: dispatchState,
|
|
ExecutionError: &evidence,
|
|
ErrorSummary: evidence.Summary,
|
|
Usage: usage.Result{State: usage.StateNotApplicable},
|
|
}
|
|
}
|
|
|
|
func (forwarder *ExecutionForwarder) ForwardStream(
|
|
ctx context.Context,
|
|
input ForwardInput,
|
|
downstream http.ResponseWriter,
|
|
) UpstreamResult {
|
|
spec, err := newExecutionAttemptSpec(input)
|
|
if err != nil || forwarder == nil || forwarder.executor == nil || downstream == nil {
|
|
return executionInputFailure(err)
|
|
}
|
|
|
|
writeTimeout := forwarder.writeTimeout
|
|
if writeTimeout <= 0 {
|
|
writeTimeout = downstreamWriteTimeout
|
|
}
|
|
controller := newStreamWriteController(downstream, writeTimeout)
|
|
defer func() { _ = controller.clear() }()
|
|
usageCapture := forwarder.usageCapture
|
|
if usageCapture == nil {
|
|
usageCapture = newUsageCaptureBoundary()
|
|
}
|
|
streamEvents := newStreamEventObserver(
|
|
input.Dialect,
|
|
usageCapture.newStreamForRequest(input.Dialect, input.ObserveUsage),
|
|
)
|
|
redactor := redact.New()
|
|
if forwarder.representation != nil && forwarder.representation.redactor != nil {
|
|
redactor = forwarder.representation.redactor
|
|
}
|
|
summarySecrets := resolvedErrorSummarySecretValues(
|
|
input.APIKey,
|
|
input.Group.HeaderRules,
|
|
input.CredentialSecrets...,
|
|
)
|
|
credentialSecrets := append([]string(nil), input.CredentialSecrets...)
|
|
if input.APIKey != "" {
|
|
credentialSecrets = append(credentialSecrets, input.APIKey)
|
|
}
|
|
streamBuffer := newSSEEventObservationBuffer(execution.SSEEventLimit(input.ClientProtocol), func(
|
|
event dialect.StreamEvent,
|
|
genericProviderError bool,
|
|
) (bool, error) {
|
|
if input.ClientProtocol == protocol.OpenAIImages &&
|
|
imagesCredentialLiteralsRemain(event.Payload, credentialSecrets) {
|
|
return false, fmt.Errorf("%w: credential remains in response event", ErrUpstreamProtocol)
|
|
}
|
|
wasTerminal := streamEvents.sawTerminal
|
|
providerError, err := streamEvents.classify(event, genericProviderError)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if !wasTerminal && !providerError && input.OnResponse != nil {
|
|
object, err := decodeResponsesStoreObject(event.Payload)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if response, exists := object["response"]; exists {
|
|
if err := input.OnResponse(response); err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
}
|
|
if !wasTerminal {
|
|
streamEvents.observeUsageEvent(event)
|
|
if providerError {
|
|
streamEvents.observeError(
|
|
event.Payload,
|
|
summarizeErrorBody(
|
|
redactor,
|
|
event.Payload,
|
|
fixedErrorSummary("upstream_sse_error"),
|
|
summarySecrets...,
|
|
),
|
|
)
|
|
}
|
|
}
|
|
return !wasTerminal && streamEvents.sawTerminal, nil
|
|
})
|
|
var responsesStoreBuffer *statelessResponsesSSEBuffer
|
|
if input.ResponsesStoreDowngraded {
|
|
responsesStoreBuffer = newStatelessResponsesSSEBuffer()
|
|
}
|
|
|
|
var (
|
|
ready *execution.StreamEvent
|
|
committed bool
|
|
firstResponse bool
|
|
downstreamErr error
|
|
errorBody []byte
|
|
streamUsage *execution.UsageEvidence
|
|
)
|
|
sink := func(event execution.StreamEvent) error {
|
|
if err := event.Validate(); err != nil {
|
|
downstreamErr = fmt.Errorf("%w: invalid execution stream event", ErrUpstreamProtocol)
|
|
return downstreamErr
|
|
}
|
|
switch event.Kind {
|
|
case execution.StreamEventReady:
|
|
copy := event.Clone()
|
|
if copy.StatusCode >= http.StatusOK && copy.StatusCode < http.StatusMultipleChoices {
|
|
if err := validateStreamContentEncoding(copy.Header); err != nil {
|
|
downstreamErr = err
|
|
return err
|
|
}
|
|
copy.Header = normalizeStreamResponseHeaders(
|
|
sanitizeForwardResponseHeaders(copy.Header, input),
|
|
)
|
|
} else {
|
|
copy.Header = sanitizeForwardResponseHeaders(copy.Header, input)
|
|
}
|
|
ready = ©
|
|
return nil
|
|
case execution.StreamEventUsage:
|
|
if event.Usage != nil {
|
|
copy := event.Usage.Clone()
|
|
streamUsage = ©
|
|
}
|
|
return nil
|
|
case execution.StreamEventData:
|
|
if ready == nil {
|
|
downstreamErr = fmt.Errorf("%w: execution data arrived before response metadata", ErrUpstreamProtocol)
|
|
return downstreamErr
|
|
}
|
|
if ready.StatusCode < http.StatusOK || ready.StatusCode >= http.StatusMultipleChoices {
|
|
errorBody = appendExecutionErrorBody(errorBody, event.Data)
|
|
return nil
|
|
}
|
|
observedData := event.Data
|
|
if responsesStoreBuffer != nil {
|
|
observedData, err = responsesStoreBuffer.push(event.Data)
|
|
if err != nil {
|
|
downstreamErr = executionStreamProtocolFailure(err)
|
|
return downstreamErr
|
|
}
|
|
if len(observedData) == 0 {
|
|
return nil
|
|
}
|
|
}
|
|
completeData, terminalInChunk, err := streamBuffer.push(observedData)
|
|
if err != nil {
|
|
downstreamErr = executionStreamProtocolFailure(err)
|
|
return downstreamErr
|
|
}
|
|
forwardData := observedData
|
|
if input.ClientProtocol == protocol.OpenAIImages || responsesStoreBuffer != nil || input.OnResponse != nil {
|
|
forwardData = completeData
|
|
if len(forwardData) == 0 {
|
|
return nil
|
|
}
|
|
}
|
|
if !committed {
|
|
if !firstResponse {
|
|
firstResponse = true
|
|
if input.OnFirstResponse != nil {
|
|
input.OnFirstResponse()
|
|
}
|
|
}
|
|
if streamEvents.firstEventWasProviderError() {
|
|
errorBody = appendExecutionErrorBody(errorBody, forwardData)
|
|
return nil
|
|
}
|
|
committed = true
|
|
if err := commitStream(controller, ready.StatusCode, ready.Header, forwardData); err != nil {
|
|
downstreamErr = err
|
|
return err
|
|
}
|
|
if input.OnStreamReady != nil {
|
|
input.OnStreamReady()
|
|
}
|
|
if terminalInChunk {
|
|
streamEvents.markTerminalForwarded()
|
|
}
|
|
return nil
|
|
}
|
|
written, err := controller.write(forwardData)
|
|
if err != nil {
|
|
downstreamErr = &streamFailure{
|
|
kind: streamFailureDownstreamWrite,
|
|
err: fmt.Errorf("write execution stream: %w", err),
|
|
}
|
|
return downstreamErr
|
|
}
|
|
if written != len(forwardData) {
|
|
downstreamErr = &streamFailure{
|
|
kind: streamFailureDownstreamWrite,
|
|
err: fmt.Errorf("write execution stream: %w", io.ErrShortWrite),
|
|
}
|
|
return downstreamErr
|
|
}
|
|
if err := controller.flush(); err != nil {
|
|
downstreamErr = &streamFailure{
|
|
kind: streamFailureDownstreamWrite,
|
|
err: fmt.Errorf("flush execution stream: %w", err),
|
|
}
|
|
return downstreamErr
|
|
}
|
|
if terminalInChunk {
|
|
streamEvents.markTerminalForwarded()
|
|
}
|
|
return nil
|
|
default:
|
|
downstreamErr = fmt.Errorf("%w: unsupported execution stream event", ErrUpstreamProtocol)
|
|
return downstreamErr
|
|
}
|
|
}
|
|
|
|
terminal := forwarder.executor.ExecuteStream(ctx, spec, sink)
|
|
if err := terminal.Validate(); err != nil {
|
|
terminal = invalidExecutionStreamResult(terminal, ready, committed)
|
|
}
|
|
if downstreamErr == nil && terminal.Error == nil {
|
|
if responsesStoreBuffer != nil {
|
|
if err := responsesStoreBuffer.finish(); err != nil {
|
|
downstreamErr = executionStreamProtocolFailure(err)
|
|
}
|
|
}
|
|
if downstreamErr == nil {
|
|
if err := streamBuffer.finish(); err != nil {
|
|
downstreamErr = executionStreamProtocolFailure(err)
|
|
} else if err := streamEvents.validateEOF(); err != nil {
|
|
downstreamErr = err
|
|
}
|
|
}
|
|
}
|
|
capturedUsage := streamEvents.finalizeUsage()
|
|
result := upstreamFromExecutionStreamResult(ctx, input, terminal, streamUsage)
|
|
if input.ObserveUsage && input.ClientProtocol == protocol.Anthropic &&
|
|
input.RouteMode == execution.RouteNative && capturedUsage.State != usage.StateMissing {
|
|
// 原生 Anthropic 流以实际事件为用量依据。SDK 的最大值合并会丢失
|
|
// 输入修正、明确的零、缓存 TTL 明细，以及流是否完整结束的信息。
|
|
result.Usage = capturedUsage
|
|
} else {
|
|
// 未采集到用量时，保留执行层已经取得的证据。
|
|
result.Usage = preferCapturedStreamUsage(result.Usage, capturedUsage)
|
|
}
|
|
result.Committed = committed
|
|
if len(errorBody) > 0 {
|
|
result.Body = append([]byte(nil), errorBody...)
|
|
result.ClassificationBody = append([]byte(nil), errorBody...)
|
|
}
|
|
if downstreamErr != nil {
|
|
result.Err = downstreamErr
|
|
}
|
|
if committed {
|
|
result.Stream = executionStreamObservation(ctx, terminal, downstreamErr, streamEvents)
|
|
} else if ready != nil {
|
|
result.StatusCode = ready.StatusCode
|
|
result.Header = ready.Header.Clone()
|
|
result.ResponseStarted = true
|
|
result.UpstreamRequestID = ready.UpstreamRequestID
|
|
}
|
|
if !committed && streamEvents.firstEventWasProviderError() {
|
|
summary := streamEvents.firstSummary
|
|
if summary == "" {
|
|
summary = fixedErrorSummary("upstream_sse_error")
|
|
}
|
|
result.ProviderErrorBeforeCommit = true
|
|
result.ErrorSummary = summary
|
|
result.ExecutionError = firstStreamErrorEvidence(
|
|
streamEvents.firstErrorPayload,
|
|
result.ExecutionError,
|
|
result.StatusCode,
|
|
result.UpstreamRequestID,
|
|
summary,
|
|
redactor,
|
|
summarySecrets,
|
|
)
|
|
}
|
|
if !committed && result.HasResponse() && !result.ProviderErrorBeforeCommit {
|
|
result = forwarder.prepareBufferedResult(input, result)
|
|
}
|
|
return result
|
|
}
|
|
|
|
func invalidExecutionStreamResult(
|
|
result execution.StreamResult,
|
|
ready *execution.StreamEvent,
|
|
committed bool,
|
|
) execution.StreamResult {
|
|
dispatchState := execution.DispatchNotSent
|
|
if result.DispatchState == execution.DispatchLocal && !committed && ready == nil {
|
|
dispatchState = execution.DispatchLocal
|
|
} else if committed || ready != nil || result.DispatchState == execution.DispatchMaybeSent ||
|
|
result.ResponseStarted || result.StatusCode != 0 || len(result.Header) > 0 {
|
|
dispatchState = execution.DispatchMaybeSent
|
|
}
|
|
normalized := execution.StreamResult{DispatchState: dispatchState}
|
|
if ready != nil {
|
|
normalized.ResponseStarted = true
|
|
normalized.StatusCode = ready.StatusCode
|
|
normalized.Header = ready.Header.Clone()
|
|
normalized.UpstreamRequestID = ready.UpstreamRequestID
|
|
}
|
|
evidence := execution.ErrorEvidence{
|
|
Kind: execution.ErrorKindInternal,
|
|
OriginHint: execution.ErrorOriginInternal,
|
|
ScopeHint: execution.ErrorScopeRequest,
|
|
Code: "attempt_result_contract_invalid",
|
|
Summary: "Attempt executor returned an invalid result.",
|
|
ReplaySafety: execution.ReplaySafetyUnknown,
|
|
}
|
|
if normalized.ResponseStarted {
|
|
evidence.StatusCode = normalized.StatusCode
|
|
}
|
|
normalized.Error = &evidence
|
|
return normalized
|
|
}
|
|
|
|
func firstStreamErrorEvidence(
|
|
payload []byte,
|
|
existing *execution.ErrorEvidence,
|
|
statusCode int,
|
|
requestID string,
|
|
summary string,
|
|
redactor *redact.Redactor,
|
|
knownSecrets []string,
|
|
) *execution.ErrorEvidence {
|
|
evidence := execution.ErrorEvidence{
|
|
Kind: execution.ErrorKindProvider,
|
|
StatusCode: statusCode,
|
|
Summary: summary,
|
|
RequestID: requestID,
|
|
}
|
|
if existing != nil {
|
|
evidence = existing.Clone()
|
|
evidence.Kind = execution.ErrorKindProvider
|
|
if evidence.StatusCode == 0 {
|
|
evidence.StatusCode = statusCode
|
|
}
|
|
if evidence.Summary == "" {
|
|
evidence.Summary = summary
|
|
}
|
|
if evidence.RequestID == "" {
|
|
evidence.RequestID = requestID
|
|
}
|
|
}
|
|
|
|
type errorObject struct {
|
|
Type string `json:"type"`
|
|
Code json.RawMessage `json:"code"`
|
|
Status string `json:"status"`
|
|
Message string `json:"message"`
|
|
}
|
|
var envelope struct {
|
|
Type string `json:"type"`
|
|
Code json.RawMessage `json:"code"`
|
|
Error errorObject `json:"error"`
|
|
Response struct {
|
|
Error errorObject `json:"error"`
|
|
} `json:"response"`
|
|
}
|
|
if json.Unmarshal(payload, &envelope) != nil {
|
|
return &evidence
|
|
}
|
|
providerError := envelope.Error
|
|
if providerError.Type == "" && len(providerError.Code) == 0 &&
|
|
providerError.Status == "" && providerError.Message == "" {
|
|
providerError = envelope.Response.Error
|
|
}
|
|
typeValue := providerError.Type
|
|
if typeValue == "" && !strings.EqualFold(envelope.Type, "error") {
|
|
typeValue = envelope.Type
|
|
}
|
|
codeValue := streamErrorEvidenceScalar(providerError.Code)
|
|
if codeValue == "" {
|
|
codeValue = streamErrorEvidenceScalar(envelope.Code)
|
|
}
|
|
if codeValue == "" {
|
|
codeValue = providerError.Status
|
|
}
|
|
typeValue = sanitizeStreamErrorEvidenceValue(redactor, typeValue, knownSecrets)
|
|
codeValue = sanitizeStreamErrorEvidenceValue(redactor, codeValue, knownSecrets)
|
|
statusValue := sanitizeStreamErrorEvidenceValue(redactor, providerError.Status, knownSecrets)
|
|
if evidence.Type == "" {
|
|
evidence.Type = typeValue
|
|
}
|
|
if evidence.Code == "" {
|
|
evidence.Code = codeValue
|
|
}
|
|
if evidence.Hint == "" {
|
|
evidence.Hint = streamErrorFailureHint(
|
|
statusCode,
|
|
typeValue,
|
|
codeValue,
|
|
statusValue,
|
|
evidence.Summary,
|
|
)
|
|
}
|
|
evidence.OriginHint = execution.ErrorOriginUpstream
|
|
switch evidence.Hint {
|
|
case execution.FailureHintInvalidCredential,
|
|
execution.FailureHintRefreshRequired,
|
|
execution.FailureHintReauthorizationRequired:
|
|
evidence.ScopeHint = execution.ErrorScopeCredential
|
|
case execution.FailureHintModelUnavailable,
|
|
execution.FailureHintCandidateUnavailable:
|
|
evidence.ScopeHint = execution.ErrorScopeModel
|
|
case execution.FailureHintRequestRejected:
|
|
evidence.ScopeHint = execution.ErrorScopeRequest
|
|
case execution.FailureHintHostError:
|
|
evidence.ScopeHint = execution.ErrorScopeGroup
|
|
}
|
|
if evidence.ReplaySafety == "" && streamBootstrapCapacityRejection(evidence.Code) {
|
|
evidence.ReplaySafety = execution.ReplaySafetyRejectedBeforeProcessing
|
|
}
|
|
return &evidence
|
|
}
|
|
|
|
func streamBootstrapCapacityRejection(codeValue string) bool {
|
|
codeValue = strings.ToLower(strings.TrimSpace(codeValue))
|
|
return codeValue == "server_is_overloaded" || codeValue == "rate_limit_exceeded"
|
|
}
|
|
|
|
func streamErrorEvidenceScalar(value json.RawMessage) string {
|
|
value = json.RawMessage(strings.TrimSpace(string(value)))
|
|
if len(value) == 0 || string(value) == "null" {
|
|
return ""
|
|
}
|
|
var text string
|
|
if json.Unmarshal(value, &text) == nil {
|
|
return text
|
|
}
|
|
var number json.Number
|
|
if json.Unmarshal(value, &number) == nil {
|
|
return number.String()
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func sanitizeStreamErrorEvidenceValue(
|
|
redactor *redact.Redactor,
|
|
value string,
|
|
knownSecrets []string,
|
|
) string {
|
|
value = sanitizeErrorSummary(redactor, value, knownSecrets...)
|
|
const maxEvidenceRunes = 128
|
|
runes := []rune(value)
|
|
if len(runes) > maxEvidenceRunes {
|
|
return string(runes[:maxEvidenceRunes])
|
|
}
|
|
return value
|
|
}
|
|
|
|
func streamErrorFailureHint(statusCode int, values ...string) execution.FailureHint {
|
|
markers := strings.ToLower(strings.Join(values, " "))
|
|
switch {
|
|
case statusCode == http.StatusUnauthorized:
|
|
return execution.FailureHintInvalidCredential
|
|
case containsStreamErrorMarker(markers,
|
|
"model_not_found", "model not found", "model_not_available",
|
|
"model unavailable", "deployment_not_found", "unsupported_model"):
|
|
return execution.FailureHintModelUnavailable
|
|
case statusCode == http.StatusTooManyRequests || containsStreamErrorMarker(markers,
|
|
"rate_limit", "rate limit", "too_many_requests", "quota_exceeded",
|
|
"resource_exhausted", "throttl"):
|
|
return execution.FailureHintRateLimited
|
|
case containsStreamErrorMarker(markers,
|
|
"invalid_api_key", "api_key_invalid", "authentication_error",
|
|
"authentication failed",
|
|
"invalid credential", "api key not valid"):
|
|
return execution.FailureHintInvalidCredential
|
|
case statusCode >= http.StatusInternalServerError && statusCode <= 599:
|
|
return execution.FailureHintHostError
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
func containsStreamErrorMarker(value string, markers ...string) bool {
|
|
for _, marker := range markers {
|
|
if strings.Contains(value, marker) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (forwarder *ExecutionForwarder) prepareBufferedResult(
|
|
input ForwardInput,
|
|
result UpstreamResult,
|
|
) UpstreamResult {
|
|
if forwarder == nil || !result.HasResponse() {
|
|
return result
|
|
}
|
|
if input.Operation == execution.OperationWebSearch && result.ProviderErrorBeforeCommit {
|
|
return result
|
|
}
|
|
representation := forwarder.representation
|
|
if representation == nil {
|
|
representation = &responseProcessor{redactor: redact.New()}
|
|
}
|
|
secrets := append([]string(nil), input.CredentialSecrets...)
|
|
if input.APIKey != "" {
|
|
secrets = append(secrets, input.APIKey)
|
|
}
|
|
if result.StatusCode >= http.StatusOK && result.StatusCode < http.StatusMultipleChoices {
|
|
policy := classifyResponseBody(input.Request.Method, result.StatusCode)
|
|
if !policy.readBody {
|
|
if err := validateStreamContentEncoding(result.Header); err != nil ||
|
|
(result.StatusCode == http.StatusPartialContent && !hasSafeHeadContentRange(result.Header)) {
|
|
return executionRepresentationFailure(result, fmt.Errorf("%w: invalid bodyless response representation", ErrUpstreamProtocol))
|
|
}
|
|
headers := sanitizeForwardResponseHeaders(result.Header, input, secrets...)
|
|
headers, _ = normalizeBufferedResponse(input.Request.Method, result.StatusCode, headers, nil)
|
|
result.Header = headers
|
|
result.Body = nil
|
|
return result
|
|
}
|
|
prepared, err := representation.prepareSuccessRepresentation(
|
|
input, result.StatusCode, result.Header, result.Body, secrets,
|
|
)
|
|
if err != nil {
|
|
return executionRepresentationFailure(result, err)
|
|
}
|
|
result.Header = prepared.headers
|
|
result.Body = prepared.downstream
|
|
result.ClassificationBody = prepared.inspectable
|
|
return result
|
|
}
|
|
if result.ExecutionError != nil &&
|
|
(input.RouteMode == execution.RouteConverted || len(result.Body) == 0) {
|
|
result.Body = encodeClientErrorBody(input.ClientProtocol, result.StatusCode, *result.ExecutionError)
|
|
result.Header = result.Header.Clone()
|
|
if result.Header == nil {
|
|
result.Header = make(http.Header)
|
|
}
|
|
deleteHeaderField(result.Header, "Content-Encoding")
|
|
deleteHeaderField(result.Header, "Content-Length")
|
|
invalidateRewrittenBodyHeaders(result.Header)
|
|
result.Header.Set("Content-Type", "application/json")
|
|
}
|
|
prepared := representation.prepareErrorRepresentation(input, result.Header, result.Body, secrets)
|
|
result.Header = prepared.headers
|
|
result.Body = prepared.downstream
|
|
result.ClassificationBody = prepared.classification
|
|
return result
|
|
}
|
|
|
|
func encodeClientErrorBody(
|
|
clientProtocol protocol.Protocol,
|
|
status int,
|
|
evidence execution.ErrorEvidence,
|
|
) []byte {
|
|
message := strings.TrimSpace(evidence.Summary)
|
|
if message == "" {
|
|
message = "upstream request failed"
|
|
}
|
|
typeValue := strings.TrimSpace(evidence.Type)
|
|
if typeValue == "" {
|
|
switch evidence.Hint {
|
|
case execution.FailureHintInvalidCredential:
|
|
typeValue = "authentication_error"
|
|
case execution.FailureHintRateLimited:
|
|
typeValue = "rate_limit_error"
|
|
case execution.FailureHintModelUnavailable:
|
|
typeValue = "model_not_found"
|
|
default:
|
|
typeValue = "api_error"
|
|
}
|
|
}
|
|
codeValue := strings.TrimSpace(evidence.Code)
|
|
|
|
var value any
|
|
switch clientProtocol {
|
|
case protocol.Anthropic:
|
|
payload := struct {
|
|
Type string `json:"type"`
|
|
Error struct {
|
|
Type string `json:"type"`
|
|
Message string `json:"message"`
|
|
} `json:"error"`
|
|
}{Type: "error"}
|
|
payload.Error.Type = typeValue
|
|
payload.Error.Message = message
|
|
value = payload
|
|
case protocol.Gemini:
|
|
if status < 100 || status > 599 {
|
|
status = http.StatusBadGateway
|
|
}
|
|
if codeValue == "" {
|
|
codeValue = strings.ToUpper(typeValue)
|
|
}
|
|
payload := struct {
|
|
Error struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
Status string `json:"status"`
|
|
} `json:"error"`
|
|
}{}
|
|
payload.Error.Code = status
|
|
payload.Error.Message = message
|
|
payload.Error.Status = codeValue
|
|
value = payload
|
|
default:
|
|
payload := struct {
|
|
Error struct {
|
|
Message string `json:"message"`
|
|
Type string `json:"type"`
|
|
Code string `json:"code,omitempty"`
|
|
} `json:"error"`
|
|
}{}
|
|
payload.Error.Message = message
|
|
payload.Error.Type = typeValue
|
|
payload.Error.Code = codeValue
|
|
value = payload
|
|
}
|
|
body, err := json.Marshal(value)
|
|
if err != nil {
|
|
return []byte(`{"error":{"message":"upstream request failed","type":"api_error"}}`)
|
|
}
|
|
return body
|
|
}
|
|
|
|
func executionRepresentationFailure(result UpstreamResult, err error) UpstreamResult {
|
|
evidence := execution.ErrorEvidence{
|
|
Kind: execution.ErrorKindInternal, OriginHint: execution.ErrorOriginInternal,
|
|
ScopeHint: execution.ErrorScopeRequest, Code: "response_representation_invalid",
|
|
Summary: "Upstream response representation could not be processed.",
|
|
ReplaySafety: execution.ReplaySafetyUnknown,
|
|
}
|
|
if result.ResponseStarted {
|
|
evidence.StatusCode = result.StatusCode
|
|
}
|
|
return UpstreamResult{
|
|
Err: err,
|
|
StatusCode: result.StatusCode,
|
|
RequestWritten: result.RequestWritten,
|
|
DispatchState: result.DispatchState,
|
|
ResponseStarted: result.ResponseStarted,
|
|
UpstreamProtocol: result.UpstreamProtocol,
|
|
AppliedReasoning: result.AppliedReasoning.Clone(),
|
|
UpstreamRequestID: result.UpstreamRequestID,
|
|
ExecutionError: &evidence,
|
|
ErrorSummary: evidence.Summary,
|
|
}
|
|
}
|
|
|
|
func newExecutionAttemptSpec(input ForwardInput) (execution.AttemptSpec, error) {
|
|
if input.Request == nil {
|
|
return execution.AttemptSpec{}, fmt.Errorf("request is required")
|
|
}
|
|
headers := cloneEndToEndHeaders(input.Request.Header)
|
|
removeDownstreamCredentials(headers)
|
|
for name, value := range input.Group.HeaderRules.Set {
|
|
headers.Set(name, strings.ReplaceAll(value, "${API_KEY}", input.APIKey))
|
|
}
|
|
for _, name := range input.Group.HeaderRules.Remove {
|
|
headers.Del(name)
|
|
}
|
|
sanitizeUpstreamRequestHeaders(headers)
|
|
headers.Set("Accept-Encoding", "identity")
|
|
spec := execution.NewAttemptSpec(execution.AttemptSpec{
|
|
RequestID: input.RequestID,
|
|
AttemptID: input.AttemptID,
|
|
Sequence: input.AttemptSequence,
|
|
ChannelID: input.ChannelID,
|
|
RouteMode: input.RouteMode,
|
|
ClientProtocol: input.ClientProtocol,
|
|
Operation: input.Operation,
|
|
RouteRequirement: input.RouteRequirement,
|
|
ResponsesStorePreference: input.ResponsesStorePreference,
|
|
ResponsesStoreDowngraded: input.ResponsesStoreDowngraded,
|
|
ClientModel: input.ExternalModel,
|
|
UpstreamModel: input.UpstreamModelID,
|
|
Method: input.Request.Method,
|
|
Path: input.Request.Path,
|
|
RawQuery: input.Request.RawQuery,
|
|
Header: headers,
|
|
ConfiguredHeaders: input.Group.HeaderRules.ConfiguredNames(),
|
|
Body: input.Request.Body,
|
|
IncludeUsage: input.ObserveUsage,
|
|
ForceCredentialRefresh: input.ForceCredentialRefresh,
|
|
ContinuityKey: input.ContinuityKey,
|
|
TargetConfig: input.TargetConfig,
|
|
Timeouts: execution.AttemptTimeouts{
|
|
FirstByte: input.Group.Timeouts.FirstByte,
|
|
Request: input.Group.Timeouts.Request,
|
|
StreamIdle: input.Group.Timeouts.StreamIdle,
|
|
},
|
|
Credential: input.Credential,
|
|
Proxy: input.Proxy,
|
|
ProxyFingerprint: input.ProxyFingerprint,
|
|
})
|
|
if spec.ResponsesStoreDowngraded {
|
|
body, err := forceStatelessResponsesRequest(spec.Body)
|
|
if err != nil {
|
|
return execution.AttemptSpec{}, err
|
|
}
|
|
spec.Body = body
|
|
platformheader.StripRepresentationMetadata(spec.Header)
|
|
}
|
|
if err := spec.Validate(); err != nil {
|
|
return execution.AttemptSpec{}, err
|
|
}
|
|
return spec, nil
|
|
}
|
|
|
|
func upstreamFromExecutionResult(
|
|
ctx context.Context,
|
|
input ForwardInput,
|
|
result execution.AttemptResult,
|
|
) UpstreamResult {
|
|
upstream := baseExecutionResult(input, result.DispatchState, result.ResponseStarted,
|
|
result.StatusCode, result.Header, result.Model, result.UpstreamRequestID,
|
|
result.Usage, result.Error)
|
|
if result.AppliedReasoning != nil {
|
|
upstream.AppliedReasoning = result.AppliedReasoning.Clone()
|
|
}
|
|
upstream.UpstreamProtocol = result.UpstreamProtocol
|
|
if input.ClientProtocol == protocol.OpenAIImages ||
|
|
input.ClientProtocol == protocol.OpenAIEmbeddings {
|
|
// AttemptResult owns Body after the executor returns. Buffered opaque
|
|
// representations consume it synchronously, so move that ownership across
|
|
// the internal boundary instead of cloning a large payload twice.
|
|
upstream.Body = result.Body
|
|
upstream.ClassificationBody = result.Body
|
|
} else {
|
|
upstream.Body = append([]byte(nil), result.Body...)
|
|
upstream.ClassificationBody = append([]byte(nil), result.Body...)
|
|
}
|
|
if !result.ResponseStarted && result.Error != nil {
|
|
upstream.Err = executionFailureError(ctx, result.Error)
|
|
}
|
|
return upstream
|
|
}
|
|
|
|
func upstreamFromExecutionStreamResult(
|
|
ctx context.Context,
|
|
input ForwardInput,
|
|
result execution.StreamResult,
|
|
streamUsage *execution.UsageEvidence,
|
|
) UpstreamResult {
|
|
usageEvidence := result.Usage
|
|
if usageEvidence == nil {
|
|
usageEvidence = streamUsage
|
|
}
|
|
upstream := baseExecutionResult(input, result.DispatchState, result.ResponseStarted,
|
|
result.StatusCode, result.Header, result.Model, result.UpstreamRequestID,
|
|
usageEvidence, result.Error)
|
|
if result.AppliedReasoning != nil {
|
|
upstream.AppliedReasoning = result.AppliedReasoning.Clone()
|
|
}
|
|
upstream.UpstreamProtocol = result.UpstreamProtocol
|
|
if !result.ResponseStarted && result.Error != nil {
|
|
upstream.Err = executionFailureError(ctx, result.Error)
|
|
}
|
|
return upstream
|
|
}
|
|
|
|
func baseExecutionResult(
|
|
input ForwardInput,
|
|
dispatchState execution.DispatchState,
|
|
responseStarted bool,
|
|
statusCode int,
|
|
header http.Header,
|
|
model string,
|
|
upstreamRequestID string,
|
|
usageEvidence *execution.UsageEvidence,
|
|
errorEvidence *execution.ErrorEvidence,
|
|
) UpstreamResult {
|
|
result := UpstreamResult{
|
|
StatusCode: statusCode,
|
|
Header: cloneEndToEndHeaders(header),
|
|
UpstreamReportedModel: model,
|
|
ResponseModelObserved: model != "",
|
|
RequestWritten: dispatchState == execution.DispatchMaybeSent,
|
|
DispatchState: dispatchState,
|
|
ResponseStarted: responseStarted,
|
|
UpstreamRequestID: upstreamRequestID,
|
|
}
|
|
if usageEvidence != nil {
|
|
result.Usage = usageEvidence.Normalized
|
|
} else if input.ObserveUsage {
|
|
result.Usage = usage.Result{State: usage.StateMissing}
|
|
} else {
|
|
result.Usage = usage.Result{State: usage.StateNotApplicable}
|
|
}
|
|
if errorEvidence != nil {
|
|
copy := errorEvidence.Clone()
|
|
result.ExecutionError = ©
|
|
result.ErrorSummary = copy.Summary
|
|
result.ProviderErrorBeforeCommit = responseStarted &&
|
|
statusCode >= http.StatusOK && statusCode < http.StatusMultipleChoices
|
|
}
|
|
return result
|
|
}
|
|
|
|
func executionFailureError(ctx context.Context, evidence *execution.ErrorEvidence) error {
|
|
if ctx != nil && ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
if evidence == nil {
|
|
return fmt.Errorf("upstream execution failed")
|
|
}
|
|
switch evidence.Kind {
|
|
case execution.ErrorKindCanceled:
|
|
return context.Canceled
|
|
case execution.ErrorKindTimeout:
|
|
return upstreamExecutionTimeoutError{}
|
|
case execution.ErrorKindInvalidRequest, execution.ErrorKindProvider, execution.ErrorKindInternal:
|
|
return fmt.Errorf("%w: upstream execution failed", ErrUpstreamProtocol)
|
|
default:
|
|
return fmt.Errorf("upstream execution failed")
|
|
}
|
|
}
|
|
|
|
type upstreamExecutionTimeoutError struct{}
|
|
|
|
func (upstreamExecutionTimeoutError) Error() string { return "upstream execution timed out" }
|
|
func (upstreamExecutionTimeoutError) Timeout() bool { return true }
|
|
func (upstreamExecutionTimeoutError) Temporary() bool { return true }
|
|
|
|
func executionInputFailure(cause error) UpstreamResult {
|
|
if cause == nil {
|
|
cause = fmt.Errorf("execution runtime is unavailable")
|
|
}
|
|
evidence := execution.ErrorEvidence{
|
|
Kind: execution.ErrorKindInvalidRequest,
|
|
Summary: "invalid execution attempt",
|
|
}
|
|
return UpstreamResult{
|
|
Err: fmt.Errorf("%w: %v", ErrUpstreamProtocol, cause),
|
|
DispatchState: execution.DispatchNotSent,
|
|
ExecutionError: &evidence,
|
|
ErrorSummary: evidence.Summary,
|
|
}
|
|
}
|
|
|
|
func appendExecutionErrorBody(current, chunk []byte) []byte {
|
|
remaining := maxStreamingErrorBodyBytes - len(current)
|
|
if remaining <= 0 {
|
|
return current
|
|
}
|
|
if len(chunk) > remaining {
|
|
chunk = chunk[:remaining]
|
|
}
|
|
return append(current, chunk...)
|
|
}
|
|
|
|
func executionStreamProtocolFailure(cause error) error {
|
|
if cause == nil {
|
|
cause = fmt.Errorf("invalid upstream SSE stream")
|
|
}
|
|
return &streamFailure{
|
|
kind: streamFailureProtocol,
|
|
err: fmt.Errorf("%w: %v", ErrUpstreamProtocol, cause),
|
|
}
|
|
}
|
|
|
|
func preferCapturedStreamUsage(
|
|
current usage.Result,
|
|
captured usage.Result,
|
|
) usage.Result {
|
|
switch captured.State {
|
|
case usage.StateComplete:
|
|
if current.State != usage.StateComplete {
|
|
return captured
|
|
}
|
|
case usage.StatePartial:
|
|
if current.State == usage.StateMissing || current.State == "" {
|
|
return captured
|
|
}
|
|
case usage.StateMissing:
|
|
if current.State == "" {
|
|
return captured
|
|
}
|
|
}
|
|
return current
|
|
}
|
|
|
|
func executionStreamObservation(
|
|
ctx context.Context,
|
|
terminal execution.StreamResult,
|
|
downstreamErr error,
|
|
streamEvents *streamEventObserver,
|
|
) StreamObservation {
|
|
if downstreamErr != nil {
|
|
return observeStreamTermination(ctx, downstreamErr, streamEvents)
|
|
}
|
|
if terminal.Error == nil {
|
|
return streamEvents.endObservation()
|
|
}
|
|
switch terminal.Error.Kind {
|
|
case execution.ErrorKindCanceled:
|
|
return observeStreamTermination(ctx, context.Canceled, streamEvents)
|
|
case execution.ErrorKindTimeout:
|
|
return streamTerminalObservation(StreamEndIdleTimeout)
|
|
case execution.ErrorKindProvider, execution.ErrorKindHTTP:
|
|
observation := streamEvents.endObservation()
|
|
if observation.EndReason == StreamEndSSEError {
|
|
return observation
|
|
}
|
|
return StreamObservation{
|
|
EndReason: StreamEndSSEError,
|
|
ErrorSummary: terminal.Error.Summary,
|
|
}
|
|
case execution.ErrorKindInvalidRequest, execution.ErrorKindInternal:
|
|
return streamTerminalObservation(StreamEndUpstreamProtocolError)
|
|
default:
|
|
return streamTerminalObservation(StreamEndUpstreamTerminated)
|
|
}
|
|
}
|
|
|
|
var _ AttemptForwarder = (*ExecutionForwarder)(nil)
|