diff --git a/backend/internal/service/openai_capacity_shed_test.go b/backend/internal/service/openai_capacity_shed_test.go index ac32fa5749..537ec435c5 100644 --- a/backend/internal/service/openai_capacity_shed_test.go +++ b/backend/internal/service/openai_capacity_shed_test.go @@ -298,6 +298,59 @@ func TestOpenAIStreamCapacityShedAfterOutputRewritesCodeForClient(t *testing.T) require.True(t, logSink.ContainsFieldValue("upstream_request_id", "rid-shed-after-output")) } +func TestOpenAIStreamProcessingFailureAfterOutputIsRecorded(t *testing.T) { + gin.SetMode(gin.TestMode) + stream := strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_processing_failure"}}`, + "", + "event: response.output_text.delta", + `data: {"type":"response.output_text.delta","delta":"partial"}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_processing_failure","status":"failed","error":{"code":"server_error","message":"An error occurred while processing your request. Please include the request ID rid-processing-failure in your message."}}}`, + "", + }, "\n") + + for _, passthrough := range []bool{false, true} { + t.Run(map[bool]string{false: "native", true: "passthrough"}[passthrough], func(t *testing.T) { + svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}}} + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(stream)), + Header: http.Header{"X-Request-Id": []string{"rid-processing-failure"}}, + } + account := &Account{ID: 42, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Name: "codex-account"} + + var err error + if passthrough { + _, err = svc.handleStreamingResponsePassthrough(c.Request.Context(), resp, c, account, time.Now(), "gpt-5.6-terra", "gpt-5.6-terra") + } else { + _, err = svc.handleStreamingResponse(c.Request.Context(), resp, c, account, time.Now(), "gpt-5.6-terra", "gpt-5.6-terra") + } + + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr)) + require.NotEmpty(t, rec.Body.String()) + require.Contains(t, rec.Body.String(), "response.failed") + require.NotNil(t, c) + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := rawEvents.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.NotEmpty(t, events) + event := events[len(events)-1] + require.Equal(t, "stream_failed", event.Kind) + require.Equal(t, "rid-processing-failure", event.UpstreamRequestID) + require.Contains(t, event.Message, "An error occurred while processing your request") + }) + } +} + // helper 单测:只有降载码被改写,其余错误码(尤其 rate_limit_exceeded,客户端 // 依赖其原码解析重试延时)必须原样保留。 func TestSanitizeOpenAICapacityShedErrorCodeForClient(t *testing.T) { diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 7a949209b9..dfebb47a50 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -2079,6 +2079,11 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( s.handleOpenAIStreamTerminalAccountSideEffects(c, account, dataBytes, failedMessage, resp.Header, mappedModel) bareErrorAccountSideEffectsPending = false } + if eventType == "response.failed" { + // The stream cannot be replayed after semantic output. Preserve the + // terminal event, while making the upstream failure queryable. + s.recordOpenAIStreamUpstreamError(c, account, true, upstreamRequestID, "stream_failed", dataBytes, failedMessage) + } } if !outputStarted { shouldFailover := false diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 4ee9dca2b1..d5b35e8585 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -561,6 +561,12 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. s.handleOpenAIStreamTerminalAccountSideEffects(c, account, dataBytes, failedMessage, resp.Header, mappedModel) bareErrorAccountSideEffectsPending = false } + if eventType == "response.failed" { + // Once semantic output is committed, failover replay is unsafe. Keep + // the terminal event on the existing stream, but retain the upstream + // request ID and payload for operations diagnostics. + s.recordOpenAIStreamUpstreamError(c, account, false, upstreamRequestID, "stream_failed", dataBytes, failedMessage) + } } if !outputStarted { shouldFailover := false