diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 92ee773166..c2f51795c8 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -348,7 +348,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( result, handleErr = s.handleAnthropicStreamingResponse(resp, c, account, originalModel, billingModel, upstreamModel, startTime) } else { // Client wants JSON: buffer the streaming response and assemble a JSON reply. - result, handleErr = s.handleAnthropicBufferedStreamingResponse(resp, c, originalModel, billingModel, upstreamModel, startTime) + result, handleErr = s.handleAnthropicBufferedStreamingResponse(resp, c, account, originalModel, billingModel, upstreamModel, startTime) } // cyber_policy：标记已设、error 已按 Anthropic 格式发给客户端。丢弃 result、返回哨兵， @@ -424,6 +424,7 @@ func (s *OpenAIGatewayService) handleAnthropicErrorResponse( func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( resp *http.Response, c *gin.Context, + account *Account, originalModel string, billingModel string, upstreamModel string, @@ -441,8 +442,6 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( return nil, fmt.Errorf("upstream stream ended without terminal event") } - // cyber_policy：上游硬阻断（response.failed）。anthropic buffered 原对 failed 无特殊分支， - // 此处仅为 cyber 增加：以 Anthropic 错误格式回写，标记供 handler 事后写风控/邮件/tokens=0 用量行。 if strings.TrimSpace(finalResponse.Status) == "failed" { payload, _ := json.Marshal(gin.H{"type": "response.failed", "response": finalResponse}) if hit, code, msg := detectOpenAICyberPolicy(payload); hit { @@ -461,6 +460,13 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( writeAnthropicError(c, http.StatusBadRequest, "invalid_request_error", clientMsg) return nil, fmt.Errorf("openai cyber_policy: %s", msg) } + message := openAICompatFailedResponseMessage(finalResponse) + if openAIStreamFailedEventShouldFailover(payload, message) { + return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message) + } + message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) + writeAnthropicError(c, http.StatusBadGateway, "api_error", message) + return nil, fmt.Errorf("upstream response failed: %s", message) } // When the terminal event has an empty output array, reconstruct from @@ -701,6 +707,8 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( firstChunk := true clientDisconnected := false clientOutputStarted := false + var streamFailoverErr error + var streamNonFailoverErr error scanner := s.newUpstreamSSEScanner(resp.Body) @@ -767,7 +775,8 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( // cyber_policy 致命不可重试：标记供 handler 事后记录；以 Anthropic SSE error 事件 // 回写让客户端感知并停止重试（F4），丢弃后续转换输出。 if strings.TrimSpace(event.Type) == "response.failed" { - if hit, code, msg := detectOpenAICyberPolicy([]byte(payload)); hit { + payloadBytes := []byte(payload) + if hit, code, msg := detectOpenAICyberPolicy(payloadBytes); hit { MarkOpsCyberPolicy(c, CyberPolicyMark{ Code: code, Message: msg, @@ -789,6 +798,25 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( } return true } + message := extractOpenAISSEErrorMessage(payloadBytes) + if openAIStreamFailedEventShouldFailover(payloadBytes, message) { + streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message) + return true + } + message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) + if !clientDisconnected { + if !clientOutputStarted { + writeAnthropicError(c, http.StatusBadGateway, "api_error", message) + clientOutputStarted = true + } else { + writeStreamHeaders() + if _, err := fmt.Fprint(c.Writer, buildAnthropicStreamErrorSSE("api_error", message)); err == nil { + c.Writer.Flush() + } + } + } + streamNonFailoverErr = fmt.Errorf("upstream response failed: %s", message) + return true } } @@ -823,6 +851,12 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( // finalizeStream sends any remaining Anthropic events and returns the result. finalizeStream := func() (*OpenAIForwardResult, error) { + if streamFailoverErr != nil { + return resultWithUsage(), streamFailoverErr + } + if streamNonFailoverErr != nil { + return resultWithUsage(), streamNonFailoverErr + } if finalEvents := apicompat.FinalizeResponsesAnthropicStream(state); len(finalEvents) > 0 && !clientDisconnected { for _, evt := range finalEvents { sse, err := apicompat.ResponsesAnthropicEventToSSE(evt) diff --git a/backend/internal/service/openai_gateway_messages_failed_response_test.go b/backend/internal/service/openai_gateway_messages_failed_response_test.go new file mode 100644 index 0000000000..a03efca42d --- /dev/null +++ b/backend/internal/service/openai_gateway_messages_failed_response_test.go @@ -0,0 +1,107 @@ +//go:build unit + +package service + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func buildResponsesFailedSSEStream(errType, errorMessage string) string { + failed := fmt.Sprintf(`{"type":"response.failed","response":{"id":"resp_err","object":"response","status":"failed","error":{"type":"%s","message":"%s"},"output":[],"usage":{"input_tokens":10,"output_tokens":0,"total_tokens":10}}}`, errType, errorMessage) + return fmt.Sprintf("data: %s\n\n", failed) +} + +func TestForwardAsAnthropic_BufferedResponseFailed_ReturnsError(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":32,"messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + ssePayload := buildResponsesFailedSSEStream("invalid_request_error", "Content policy violation") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ssePayload)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + account := rawChatCompletionsTestAccount() + _, err := svc.ForwardAsAnthropic(context.Background(), c, account, body, "", "") + + require.Error(t, err, "non-cyber response.failed must return an error, not swallow as 200") + require.Contains(t, err.Error(), "upstream response failed") + require.Equal(t, http.StatusBadGateway, rec.Code, "should write 502 for non-failover failed response") +} + +func TestForwardAsAnthropic_StreamingResponseFailed_ReturnsError(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":32,"messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + ssePayload := buildResponsesFailedSSEStream("invalid_request_error", "Content policy violation") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ssePayload)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + account := rawChatCompletionsTestAccount() + _, err := svc.ForwardAsAnthropic(context.Background(), c, account, body, "", "") + + require.Error(t, err, "streaming response.failed must return an error") + require.Contains(t, err.Error(), "upstream response failed") +} + +func TestForwardAsAnthropic_BufferedResponseFailed_Failover(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":32,"messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + ssePayload := buildResponsesFailedSSEStream("rate_limit_error", "Rate limit reached") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ssePayload)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + account := rawChatCompletionsTestAccount() + _, err := svc.ForwardAsAnthropic(context.Background(), c, account, body, "", "") + + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.True(t, errors.As(err, &failoverErr), "rate_limit_error should trigger UpstreamFailoverError for failover, got: %T: %v", err, err) +}