From 4795650d21d26b8c1db3b64dfe292fd61ef0a1d7 Mon Sep 17 00:00:00 2001 From: shaw Date: Wed, 26 Aug 2026 09:42:28 +0800 Subject: [PATCH] fix(openai): record actual upstream endpoint on errors --- backend/internal/handler/endpoint.go | 8 ++++++++ backend/internal/handler/endpoint_test.go | 10 ++++++++++ backend/internal/service/openai_gateway_cc_pipeline.go | 4 ++++ .../service/openai_gateway_chat_completions.go | 4 ++++ backend/internal/service/openai_gateway_forward.go | 4 ++++ backend/internal/service/openai_gateway_messages.go | 4 ++++ .../openai_gateway_responses_chat_fallback_test.go | 2 ++ backend/internal/service/openai_gateway_service.go | 10 ++++++++++ 8 files changed, 46 insertions(+) diff --git a/backend/internal/handler/endpoint.go b/backend/internal/handler/endpoint.go index 6985392926..d13bbee47a 100644 --- a/backend/internal/handler/endpoint.go +++ b/backend/internal/handler/endpoint.go @@ -312,6 +312,14 @@ func GetInboundEndpoint(c *gin.Context) string { // and the account platform. Handlers call this after scheduling an // account, passing account.Platform. func GetUpstreamEndpoint(c *gin.Context, platform string) string { + // OpenAI 转发服务维护独立的运行时端点上下文，覆盖普通入站推导。 + // 这对 force_chat_completions 的错误路径尤为重要：此时可能没有 + // ForwardResult，不能把入站 /v1/responses 误报成上游端点。 + if platform == service.PlatformOpenAI || platform == service.PlatformGrok || service.IsCNProvider(platform) { + if endpoint := service.GetActualOpenAIUpstreamEndpoint(c); endpoint != "" { + return endpoint + } + } if c != nil { if value, ok := c.Get(ctxKeyActualUpstreamEndpoint); ok { if endpoint, ok := value.(string); ok && endpoint != "" { diff --git a/backend/internal/handler/endpoint_test.go b/backend/internal/handler/endpoint_test.go index 005c397d44..f87b27fcd1 100644 --- a/backend/internal/handler/endpoint_test.go +++ b/backend/internal/handler/endpoint_test.go @@ -184,6 +184,16 @@ func TestGetUpstreamEndpointPrefersRuntimeOverride(t *testing.T) { require.Equal(t, EndpointMessages, GetUpstreamEndpoint(c, service.PlatformAntigravity)) } +func TestGetUpstreamEndpointUsesOpenAIRuntimeOverride(t *testing.T) { + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil) + c.Set(ctxKeyInboundEndpoint, EndpointResponses) + + service.SetActualOpenAIUpstreamEndpoint(c, EndpointChatCompletions) + require.Equal(t, EndpointChatCompletions, GetUpstreamEndpoint(c, service.PlatformOpenAI)) +} + func TestResolveOpenAIUpstreamEndpointPrefersForwardResult(t *testing.T) { tests := []struct { name string diff --git a/backend/internal/service/openai_gateway_cc_pipeline.go b/backend/internal/service/openai_gateway_cc_pipeline.go index b0e98e5926..a22288b70a 100644 --- a/backend/internal/service/openai_gateway_cc_pipeline.go +++ b/backend/internal/service/openai_gateway_cc_pipeline.go @@ -186,6 +186,10 @@ func (s *OpenAIGatewayService) sendCCUpstreamRequest( if err != nil { return nil, fmt.Errorf("build upstream request: %w", err) } + // 记录本次实际选择的协议端点，供错误日志和用量日志在没有 + // OpenAIForwardResult（例如 503/传输失败）时使用。每次发送都覆盖， + // 避免 Gin context 在账号 failover 尝试之间残留旧端点。 + SetActualOpenAIUpstreamEndpoint(c, "/v1/chat/completions") upstreamReq = upstreamReq.WithContext(WithHTTPUpstreamProfile(upstreamReq.Context(), HTTPUpstreamProfileOpenAI)) upstreamReq.Header.Set("Content-Type", "application/json") upstreamReq.Header.Set("Authorization", "Bearer "+bearerToken) diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 69c1458fdc..a3a992175c 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -60,6 +60,10 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions( defaultMappedModel string, ) (*OpenAIForwardResult, error) { beginUpstreamResponseModelObservation(c) + ClearActualOpenAIUpstreamEndpoint(c) + if shouldForwardOpenAIResponsesViaRawChatCompletions(account) { + SetActualOpenAIUpstreamEndpoint(c, "/v1/chat/completions") + } setCodexToolNameReverse(c, nil) if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { return nil, err diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 6487385000..8dc048850c 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -20,6 +20,10 @@ import ( // Forward forwards request to OpenAI API func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, account *Account, body []byte) (*OpenAIForwardResult, error) { beginUpstreamResponseModelObservation(c) + ClearActualOpenAIUpstreamEndpoint(c) + if shouldForwardOpenAIResponsesViaRawChatCompletions(account) { + SetActualOpenAIUpstreamEndpoint(c, "/v1/chat/completions") + } clearGrokResponsesClientToolMapping(c) clearOpenAIResponsesClientToolMapping(c) clearOpenAIResponsesNamespaceNames(c) diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 17c4d633d1..a494be8adf 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -34,6 +34,10 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( defaultMappedModel string, ) (*OpenAIForwardResult, error) { beginUpstreamResponseModelObservation(c) + ClearActualOpenAIUpstreamEndpoint(c) + if shouldForwardOpenAIResponsesViaRawChatCompletions(account) { + SetActualOpenAIUpstreamEndpoint(c, "/v1/chat/completions") + } setCodexToolNameReverse(c, nil) if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { return nil, err diff --git a/backend/internal/service/openai_gateway_responses_chat_fallback_test.go b/backend/internal/service/openai_gateway_responses_chat_fallback_test.go index 750c45e8f2..7b1eb9cba8 100644 --- a/backend/internal/service/openai_gateway_responses_chat_fallback_test.go +++ b/backend/internal/service/openai_gateway_responses_chat_fallback_test.go @@ -39,11 +39,13 @@ func TestForwardResponses_ForceChatCompletionsRoutesNonStreamingToChatCompletion cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } + SetActualOpenAIUpstreamEndpoint(c, "/v1/responses") result, err := svc.Forward(context.Background(), c, forceChatResponsesFallbackAccount(), body) require.NoError(t, err) require.NotNil(t, result) require.Equal(t, "http://upstream.example/v1/chat/completions", upstream.lastReq.URL.String()) + require.Equal(t, "/v1/chat/completions", GetActualOpenAIUpstreamEndpoint(c)) require.Equal(t, HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileFromContext(upstream.lastReq.Context())) require.Equal(t, "hello", gjson.GetBytes(upstream.lastBody, "messages.0.content").String()) require.False(t, gjson.GetBytes(upstream.lastBody, "input").Exists()) diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index c4d085b024..5ce98fba0f 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -319,6 +319,16 @@ func SetActualOpenAIUpstreamEndpoint(c *gin.Context, endpoint string) { } } +// ClearActualOpenAIUpstreamEndpoint 清理当前转发尝试记录的端点。 +// Handler 会在账号 failover 尝试间复用同一个 Gin context，因此每次尝试 +// 都必须从无残留状态开始。 +func ClearActualOpenAIUpstreamEndpoint(c *gin.Context) { + if c == nil { + return + } + c.Set(openAIUpstreamEndpointContextKey, "") +} + // GetActualOpenAIUpstreamEndpoint returns the endpoint recorded by the latest // forwarding attempt in this request. func GetActualOpenAIUpstreamEndpoint(c *gin.Context) string {