diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index e135eead9c..2afd155646 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -16,6 +16,7 @@ import ( "github.com/Wei-Shaw/sub2api/internal/pkg/apicompat" "github.com/Wei-Shaw/sub2api/internal/pkg/claude" "github.com/Wei-Shaw/sub2api/internal/pkg/logger" + "github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat" "github.com/Wei-Shaw/sub2api/internal/pkg/xai" "github.com/Wei-Shaw/sub2api/internal/util/responseheaders" "github.com/gin-gonic/gin" @@ -34,6 +35,14 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( promptCacheKey string, defaultMappedModel string, ) (*OpenAIForwardResult, error) { + // 入口分流:APIKey 账号 + 上游不支持 Responses API → 走 CC 直转(与 + // ForwardAsChatCompletions 对称)。缺少此分流时,/v1/messages 入站请求 + // 会被无条件转为 Responses 格式发往上游 /v1/responses,导致只支持 + // /v1/chat/completions 的第三方 OpenAI 兼容上游全部 400。 + if account.Type == AccountTypeAPIKey && !openai_compat.ShouldUseResponsesAPI(account.Extra) { + return s.forwardAnthropicViaRawChatCompletions(ctx, c, account, body, defaultMappedModel) + } + startTime := time.Now() // 1. Parse Anthropic request diff --git a/backend/internal/service/openai_gateway_messages_chat_fallback.go b/backend/internal/service/openai_gateway_messages_chat_fallback.go new file mode 100644 index 0000000000..c9381130b7 --- /dev/null +++ b/backend/internal/service/openai_gateway_messages_chat_fallback.go @@ -0,0 +1,407 @@ +package service + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/apicompat" + "github.com/Wei-Shaw/sub2api/internal/pkg/logger" + "github.com/Wei-Shaw/sub2api/internal/util/responseheaders" + "github.com/gin-gonic/gin" + "go.uber.org/zap" +) + +// forwardAnthropicViaRawChatCompletions serves /v1/messages clients through +// an OpenAI-compatible upstream that only supports /v1/chat/completions. +// +// Conversion chain: +// +// Request: Anthropic Messages → Responses (AnthropicToResponses) +// → Chat Completions (ResponsesToChatCompletionsRequest) +// Response: CC chunk → Responses events (ChatCompletionsChunkToResponsesEvents) +// → Anthropic events (ResponsesEventToAnthropicEvents) +// +// This is the /v1/messages counterpart of forwardResponsesViaRawChatCompletions +// (which serves /v1/responses clients). The same conversion bridges are reused; +// only the inbound/outbound framing differs. +func (s *OpenAIGatewayService) forwardAnthropicViaRawChatCompletions( + ctx context.Context, + c *gin.Context, + account *Account, + body []byte, + defaultMappedModel string, +) (*OpenAIForwardResult, error) { + startTime := time.Now() + + // 1. Parse Anthropic request + var anthropicReq apicompat.AnthropicRequest + if err := json.Unmarshal(body, &anthropicReq); err != nil { + writeAnthropicError(c, http.StatusBadRequest, "invalid_request_error", "Failed to parse request body") + return nil, fmt.Errorf("parse anthropic request: %w", err) + } + originalModel := anthropicReq.Model + if strings.TrimSpace(originalModel) == "" { + writeAnthropicError(c, http.StatusBadRequest, "invalid_request_error", "model is required") + return nil, fmt.Errorf("missing model in request") + } + applyOpenAICompatModelNormalization(&anthropicReq) + clientStream := anthropicReq.Stream + + // 2. Anthropic → Responses → Chat Completions + responsesReq, err := apicompat.AnthropicToResponses(&anthropicReq) + if err != nil { + writeAnthropicError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) + return nil, fmt.Errorf("convert anthropic to responses: %w", err) + } + + billingModel := resolveOpenAIForwardModel(account, anthropicReq.Model, defaultMappedModel) + upstreamModel := normalizeOpenAIModelForUpstream(account, billingModel) + responsesReq.Model = upstreamModel + + chatReq, err := apicompat.ResponsesToChatCompletionsRequest(responsesReq) + if err != nil { + writeAnthropicError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) + return nil, fmt.Errorf("convert responses to chat completions: %w", err) + } + chatReq.Stream = clientStream + if clientStream { + chatReq.StreamOptions = &apicompat.ChatStreamOptions{IncludeUsage: true} + } + + reasoningEffort := extractOpenAIReasoningEffortFromBody(body, originalModel) + reasoningEffort = ApplyThinkingEnabledFallback(reasoningEffort, body, billingModel) + serviceTier := extractOpenAIServiceTierFromBody(body) + + chatBody, err := json.Marshal(chatReq) + if err != nil { + return nil, fmt.Errorf("marshal chat completions request: %w", err) + } + if normalizedBody, normalized := NormalizeGLMOpenAIReasoningEffort(chatBody, upstreamModel); normalized { + chatBody = normalizedBody + } + + logger.L().Debug("openai messages: forwarding via raw chat completions", + zap.Int64("account_id", account.ID), + zap.String("original_model", originalModel), + zap.String("billing_model", billingModel), + zap.String("upstream_model", upstreamModel), + zap.Bool("stream", clientStream), + ) + + // 3. Build upstream request + apiKey := account.GetOpenAIApiKey() + if apiKey == "" { + return nil, fmt.Errorf("account %d missing api_key", account.ID) + } + baseURL := account.GetOpenAIBaseURL() + if baseURL == "" { + baseURL = "https://api.openai.com" + } + validatedURL, err := s.validateUpstreamBaseURL(baseURL) + if err != nil { + return nil, fmt.Errorf("invalid base_url: %w", err) + } + targetURL := buildOpenAIChatCompletionsURL(validatedURL) + + upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx) + upstreamReq, err := http.NewRequestWithContext(upstreamCtx, http.MethodPost, targetURL, bytes.NewReader(chatBody)) + releaseUpstreamCtx() + if err != nil { + return nil, fmt.Errorf("build upstream request: %w", err) + } + upstreamReq = upstreamReq.WithContext(WithHTTPUpstreamProfile(upstreamReq.Context(), HTTPUpstreamProfileOpenAI)) + upstreamReq.Header.Set("Content-Type", "application/json") + upstreamReq.Header.Set("Authorization", "Bearer "+apiKey) + if clientStream { + upstreamReq.Header.Set("Accept", "text/event-stream") + } else { + upstreamReq.Header.Set("Accept", "application/json") + } + for key, values := range c.Request.Header { + lowerKey := strings.ToLower(key) + if openaiCCRawAllowedHeaders[lowerKey] { + for _, v := range values { + upstreamReq.Header.Add(key, v) + } + } + } + if customUA := account.GetOpenAIUserAgent(); customUA != "" { + upstreamReq.Header.Set("user-agent", customUA) + } + account.ApplyHeaderOverrides(upstreamReq.Header) + + proxyURL := "" + if account.Proxy != nil { + proxyURL = account.Proxy.URL() + } + resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency) + if err != nil { + return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false) + } + defer func() { _ = resp.Body.Close() }() + + // 4. Handle error responses + if resp.StatusCode >= 400 { + respBody := s.readUpstreamErrorBody(resp) + _ = resp.Body.Close() + resp.Body = io.NopCloser(bytes.NewReader(respBody)) + + upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody)) + upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) + if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) { + upstreamDetail := "" + if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { + maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes + if maxBytes <= 0 { + maxBytes = 2048 + } + upstreamDetail = truncateString(string(respBody), maxBytes) + } + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + Platform: account.Platform, + AccountID: account.ID, + AccountName: account.Name, + UpstreamStatusCode: resp.StatusCode, + UpstreamRequestID: resp.Header.Get("x-request-id"), + Kind: "failover", + Message: upstreamMsg, + Detail: upstreamDetail, + }) + s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel) + return nil, &UpstreamFailoverError{ + StatusCode: resp.StatusCode, + ResponseBody: respBody, + RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)), + } + } + writeAnthropicError(c, mapUpstreamStatusToAnthropicStatus(resp.StatusCode), "api_error", "Upstream error: "+upstreamMsg) + return nil, fmt.Errorf("upstream %d: %s", resp.StatusCode, upstreamMsg) + } + + // 5. Convert response + if clientStream { + return s.streamChatCompletionsAsAnthropic(c, resp, originalModel, billingModel, upstreamModel, reasoningEffort, serviceTier, startTime) + } + return s.bufferChatCompletionsAsAnthropic(c, resp, originalModel, billingModel, upstreamModel, reasoningEffort, serviceTier, startTime) +} + +func (s *OpenAIGatewayService) bufferChatCompletionsAsAnthropic( + c *gin.Context, + resp *http.Response, + originalModel string, + billingModel string, + upstreamModel string, + reasoningEffort *string, + serviceTier *string, + startTime time.Time, +) (*OpenAIForwardResult, error) { + requestID := resp.Header.Get("x-request-id") + respBody, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, openAITooLargeError) + if err != nil { + if !errors.Is(err, ErrUpstreamResponseBodyTooLarge) { + writeAnthropicError(c, http.StatusBadGateway, "api_error", "Failed to read upstream response") + } + return nil, fmt.Errorf("read upstream body: %w", err) + } + + var ccResp apicompat.ChatCompletionsResponse + if err := json.Unmarshal(respBody, &ccResp); err != nil { + writeAnthropicError(c, http.StatusBadGateway, "api_error", "Failed to parse upstream response") + return nil, fmt.Errorf("parse chat completions response: %w", err) + } + responsesResp := apicompat.ChatCompletionsResponseToResponses(&ccResp, originalModel) + + anthropicResp := apicompat.ResponsesToAnthropic(responsesResp, originalModel) + + usage := OpenAIUsage{} + if parsed, ok := extractOpenAIUsageFromJSONBytes(respBody); ok { + usage = parsed + } + + if s.responseHeaderFilter != nil { + responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) + } + c.JSON(http.StatusOK, anthropicResp) + + return &OpenAIForwardResult{ + RequestID: requestID, + Usage: usage, + Model: originalModel, + BillingModel: billingModel, + UpstreamModel: upstreamModel, + ReasoningEffort: reasoningEffort, + ServiceTier: serviceTier, + Stream: false, + Duration: time.Since(startTime), + }, nil +} + +func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic( + c *gin.Context, + resp *http.Response, + originalModel string, + billingModel string, + upstreamModel string, + reasoningEffort *string, + serviceTier *string, + startTime time.Time, +) (*OpenAIForwardResult, error) { + requestID := resp.Header.Get("x-request-id") + headersWritten := false + writeStreamHeaders := func() { + if headersWritten { + return + } + headersWritten = true + if s.responseHeaderFilter != nil { + responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) + } + c.Writer.Header().Set("Content-Type", "text/event-stream") + c.Writer.Header().Set("Cache-Control", "no-cache") + c.Writer.Header().Set("Connection", "keep-alive") + c.Writer.Header().Set("X-Accel-Buffering", "no") + c.Writer.WriteHeader(http.StatusOK) + } + + ccState := apicompat.NewChatCompletionsToResponsesStreamState(originalModel) + anthropicState := apicompat.NewResponsesEventToAnthropicState() + anthropicState.Model = originalModel + var usage OpenAIUsage + var firstTokenMs *int + clientDisconnected := false + sawDone := false + + scanner := bufio.NewScanner(resp.Body) + maxLineSize := defaultMaxLineSize + if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 { + maxLineSize = s.cfg.Gateway.MaxLineSize + } + scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize) + + for scanner.Scan() { + line := scanner.Text() + payload, ok := extractOpenAISSEDataLine(line) + if !ok { + continue + } + payload = strings.TrimSpace(payload) + if payload == "" { + continue + } + if payload == "[DONE]" { + sawDone = true + break + } + + if u := extractCCStreamUsage(payload); u != nil { + usage = *u + } + + var chunk apicompat.ChatCompletionsChunk + if err := json.Unmarshal([]byte(payload), &chunk); err != nil { + logger.L().Warn("openai messages chat fallback: failed to parse chat stream chunk", + zap.Error(err), + zap.String("request_id", requestID), + ) + continue + } + if firstTokenMs == nil && !isOpenAIChatUsageOnlyStreamChunk(payload) && chatChunkStartsResponsesOutput(&chunk) { + ms := int(time.Since(startTime).Milliseconds()) + firstTokenMs = &ms + } + + // CC chunk → Responses events → Anthropic events + responsesEvents := apicompat.ChatCompletionsChunkToResponsesEvents(&chunk, ccState) + for _, rEvent := range responsesEvents { + anthropicEvents := apicompat.ResponsesEventToAnthropicEvents(&rEvent, anthropicState) + if clientDisconnected { + continue + } + for _, aEvt := range anthropicEvents { + sse, err := apicompat.ResponsesAnthropicEventToSSE(aEvt) + if err != nil { + continue + } + writeStreamHeaders() + if _, err := fmt.Fprint(c.Writer, sse); err != nil { + clientDisconnected = true + break + } + } + } + if !clientDisconnected && len(responsesEvents) > 0 { + c.Writer.Flush() + } + } + + if err := scanner.Err(); err != nil { + if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + logger.L().Warn("openai messages chat fallback: stream read error", + zap.Error(err), + zap.String("request_id", requestID), + ) + } + } + + // Finalize CC→Responses stream (emit response.completed) + finalEvents := apicompat.FinalizeChatCompletionsResponsesStream(ccState) + for _, rEvent := range finalEvents { + if rEvent.Response != nil && rEvent.Response.Usage != nil { + usage = copyOpenAIUsageFromResponsesUsage(rEvent.Response.Usage) + } + if clientDisconnected { + continue + } + anthropicEvents := apicompat.ResponsesEventToAnthropicEvents(&rEvent, anthropicState) + for _, aEvt := range anthropicEvents { + sse, err := apicompat.ResponsesAnthropicEventToSSE(aEvt) + if err != nil { + continue + } + writeStreamHeaders() + if _, err := fmt.Fprint(c.Writer, sse); err != nil { + clientDisconnected = true + break + } + } + } + if !clientDisconnected { + c.Writer.Flush() + } + _ = sawDone + + return &OpenAIForwardResult{ + RequestID: requestID, + Usage: usage, + Model: originalModel, + BillingModel: billingModel, + UpstreamModel: upstreamModel, + ReasoningEffort: reasoningEffort, + ServiceTier: serviceTier, + Stream: true, + Duration: time.Since(startTime), + FirstTokenMs: firstTokenMs, + ClientDisconnect: clientDisconnected, + }, nil +} + +func mapUpstreamStatusToAnthropicStatus(status int) int { + switch { + case status == 401: + return http.StatusBadGateway + case status == 429: + return http.StatusTooManyRequests + case status >= 500: + return http.StatusBadGateway + default: + return status + } +}