Merge branch 'Wei-Shaw:main' into fix/open-issues-cleanup

backend/internal/handler/dto/mappers.go

@@ -523,6 +523,8 @@ func usageLogFromServiceUser(l *service.UsageLog) UsageLog {
 		Model:                 l.Model,
 		ServiceTier:           l.ServiceTier,
 		ReasoningEffort:       l.ReasoningEffort,
+		InboundEndpoint:       l.InboundEndpoint,
+		UpstreamEndpoint:      l.UpstreamEndpoint,
 		GroupID:               l.GroupID,
 		SubscriptionID:        l.SubscriptionID,
 		InputTokens:           l.InputTokens,

backend/internal/handler/dto/mappers_usage_test.go

@@ -76,10 +76,14 @@ func TestUsageLogFromService_IncludesServiceTierForUserAndAdmin(t *testing.T) {
 	t.Parallel()
 
 	serviceTier := "priority"
+	inboundEndpoint := "/v1/chat/completions"
+	upstreamEndpoint := "/v1/responses"
 	log := &service.UsageLog{
 		RequestID:             "req_3",
 		Model:                 "gpt-5.4",
 		ServiceTier:           &serviceTier,
+		InboundEndpoint:       &inboundEndpoint,
+		UpstreamEndpoint:      &upstreamEndpoint,
 		AccountRateMultiplier: f64Ptr(1.5),
 	}
 
@@ -88,8 +92,16 @@ func TestUsageLogFromService_IncludesServiceTierForUserAndAdmin(t *testing.T) {
 
 	require.NotNil(t, userDTO.ServiceTier)
 	require.Equal(t, serviceTier, *userDTO.ServiceTier)
+	require.NotNil(t, userDTO.InboundEndpoint)
+	require.Equal(t, inboundEndpoint, *userDTO.InboundEndpoint)
+	require.NotNil(t, userDTO.UpstreamEndpoint)
+	require.Equal(t, upstreamEndpoint, *userDTO.UpstreamEndpoint)
 	require.NotNil(t, adminDTO.ServiceTier)
 	require.Equal(t, serviceTier, *adminDTO.ServiceTier)
+	require.NotNil(t, adminDTO.InboundEndpoint)
+	require.Equal(t, inboundEndpoint, *adminDTO.InboundEndpoint)
+	require.NotNil(t, adminDTO.UpstreamEndpoint)
+	require.Equal(t, upstreamEndpoint, *adminDTO.UpstreamEndpoint)
 	require.NotNil(t, adminDTO.AccountRateMultiplier)
 	require.InDelta(t, 1.5, *adminDTO.AccountRateMultiplier, 1e-12)
 }

backend/internal/handler/dto/types.go

@@ -334,9 +334,13 @@ type UsageLog struct {
 	Model     string `json:"model"`
 	// ServiceTier records the OpenAI service tier used for billing, e.g. "priority" / "flex".
 	ServiceTier *string `json:"service_tier,omitempty"`
-	// ReasoningEffort is the request's reasoning effort level (OpenAI Responses API).
-	// nil means not provided / not applicable.
+	// ReasoningEffort is the request's reasoning effort level.
+	// OpenAI: "low"/"medium"/"high"/"xhigh"; Claude: "low"/"medium"/"high"/"max".
 	ReasoningEffort *string `json:"reasoning_effort,omitempty"`
+	// InboundEndpoint is the client-facing API endpoint path, e.g. /v1/chat/completions.
+	InboundEndpoint *string `json:"inbound_endpoint,omitempty"`
+	// UpstreamEndpoint is the normalized upstream endpoint path, e.g. /v1/responses.
+	UpstreamEndpoint *string `json:"upstream_endpoint,omitempty"`
 
 	GroupID        *int64 `json:"group_id"`
 	SubscriptionID *int64 `json:"subscription_id"`

backend/internal/handler/gateway_handler.go

@@ -391,6 +391,8 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
 			if fs.SwitchCount > 0 {
 				requestCtx = service.WithAccountSwitchCount(requestCtx, fs.SwitchCount, h.metadataBridgeEnabled())
 			}
+			// 记录 Forward 前已写入字节数，Forward 后若增加则说明 SSE 内容已发，禁止 failover
+			writerSizeBeforeForward := c.Writer.Size()
 			if account.Platform == service.PlatformAntigravity {
 				result, err = h.antigravityGatewayService.ForwardGemini(requestCtx, c, account, reqModel, "generateContent", reqStream, body, hasBoundSession)
 			} else {
@@ -402,6 +404,11 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
 			if err != nil {
 				var failoverErr *service.UpstreamFailoverError
 				if errors.As(err, &failoverErr) {
+					// 流式内容已写入客户端，无法撤销，禁止 failover 以防止流拼接腐化
+					if c.Writer.Size() != writerSizeBeforeForward {
+						h.handleFailoverExhausted(c, failoverErr, service.PlatformGemini, true)
+						return
+					}
 					action := fs.HandleFailoverError(c.Request.Context(), h.gatewayService, account.ID, account.Platform, failoverErr)
 					switch action {
 					case FailoverContinue:
@@ -436,6 +443,10 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
 			clientIP := ip.GetClientIP(c)
 			requestPayloadHash := service.HashUsageRequestPayload(body)
 
+			if result.ReasoningEffort == nil {
+				result.ReasoningEffort = service.NormalizeClaudeOutputEffort(parsedReq.OutputEffort)
+			}
+
 			// 使用量记录通过有界 worker 池提交，避免请求热路径创建无界 goroutine。
 			h.submitUsageRecordTask(func(ctx context.Context) {
 				if err := h.gatewayService.RecordUsage(ctx, &service.RecordUsageInput{
@@ -637,6 +648,8 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
 			if fs.SwitchCount > 0 {
 				requestCtx = service.WithAccountSwitchCount(requestCtx, fs.SwitchCount, h.metadataBridgeEnabled())
 			}
+			// 记录 Forward 前已写入字节数，Forward 后若增加则说明 SSE 内容已发，禁止 failover
+			writerSizeBeforeForward := c.Writer.Size()
 			if account.Platform == service.PlatformAntigravity && account.Type != service.AccountTypeAPIKey {
 				result, err = h.antigravityGatewayService.Forward(requestCtx, c, account, body, hasBoundSession)
 			} else {
@@ -706,6 +719,11 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
 				}
 				var failoverErr *service.UpstreamFailoverError
 				if errors.As(err, &failoverErr) {
+					// 流式内容已写入客户端，无法撤销，禁止 failover 以防止流拼接腐化
+					if c.Writer.Size() != writerSizeBeforeForward {
+						h.handleFailoverExhausted(c, failoverErr, account.Platform, true)
+						return
+					}
 					action := fs.HandleFailoverError(c.Request.Context(), h.gatewayService, account.ID, account.Platform, failoverErr)
 					switch action {
 					case FailoverContinue:
@@ -740,6 +758,10 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
 			clientIP := ip.GetClientIP(c)
 			requestPayloadHash := service.HashUsageRequestPayload(body)
 
+			if result.ReasoningEffort == nil {
+				result.ReasoningEffort = service.NormalizeClaudeOutputEffort(parsedReq.OutputEffort)
+			}
+
 			// 使用量记录通过有界 worker 池提交，避免请求热路径创建无界 goroutine。
 			h.submitUsageRecordTask(func(ctx context.Context) {
 				if err := h.gatewayService.RecordUsage(ctx, &service.RecordUsageInput{

backend/internal/handler/gateway_handler_stream_failover_test.go

@@ -0,0 +1,122 @@
+package handler
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// partialMessageStartSSE 模拟 handleStreamingResponse 已写入的首批 SSE 事件。
+const partialMessageStartSSE = "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_01\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":1}}}\n\n" +
+	"event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n"
+
+// TestStreamWrittenGuard_MessagesPath_AbortFailoverOnSSEContentWritten 验证：
+// 当 Forward 在返回 UpstreamFailoverError 前已向客户端写入 SSE 内容时，
+// 故障转移保护逻辑必须终止循环并发送 SSE 错误事件，而不是进行下一次 Forward。
+// 具体验证：
+//  1. c.Writer.Size() 检测条件正确触发（字节数已增加）
+//  2. handleFailoverExhausted 以 streamStarted=true 调用后，响应体以 SSE 错误事件结尾
+//  3. 响应体中只出现一个 message_start，不存在第二个（防止流拼接腐化）
+func TestStreamWrittenGuard_MessagesPath_AbortFailoverOnSSEContentWritten(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil)
+
+	// 步骤 1：记录 Forward 前的 writer size（模拟 writerSizeBeforeForward := c.Writer.Size()）
+	sizeBeforeForward := c.Writer.Size()
+	require.Equal(t, -1, sizeBeforeForward, "gin writer 初始 Size 应为 -1（未写入任何字节）")
+
+	// 步骤 2：模拟 Forward 已向客户端写入部分 SSE 内容（message_start + content_block_start）
+	_, err := c.Writer.Write([]byte(partialMessageStartSSE))
+	require.NoError(t, err)
+
+	// 步骤 3：验证守卫条件成立（c.Writer.Size() != sizeBeforeForward）
+	require.NotEqual(t, sizeBeforeForward, c.Writer.Size(),
+		"写入 SSE 内容后 writer size 必须增加，守卫条件应为 true")
+
+	// 步骤 4：模拟 UpstreamFailoverError（上游在流中途返回 403）
+	failoverErr := &service.UpstreamFailoverError{
+		StatusCode:   http.StatusForbidden,
+		ResponseBody: []byte(`{"error":{"type":"permission_error","message":"forbidden"}}`),
+	}
+
+	// 步骤 5：守卫触发 → 调用 handleFailoverExhausted，streamStarted=true
+	h := &GatewayHandler{}
+	h.handleFailoverExhausted(c, failoverErr, service.PlatformAnthropic, true)
+
+	body := w.Body.String()
+
+	// 断言 A：响应体中包含最初写入的 message_start SSE 事件行
+	require.Contains(t, body, "event: message_start", "响应体应包含已写入的 message_start SSE 事件")
+
+	// 断言 B：响应体以 SSE 错误事件结尾（data: {"type":"error",...}\n\n）
+	require.True(t, strings.HasSuffix(strings.TrimRight(body, "\n"), "}"),
+		"响应体应以 JSON 对象结尾（SSE error event 的 data 字段）")
+	require.Contains(t, body, `"type":"error"`, "响应体末尾必须包含 SSE 错误事件")
+
+	// 断言 C：SSE event 行 "event: message_start" 只出现一次（防止双 message_start 拼接腐化）
+	firstIdx := strings.Index(body, "event: message_start")
+	lastIdx := strings.LastIndex(body, "event: message_start")
+	assert.Equal(t, firstIdx, lastIdx,
+		"响应体中 'event: message_start' 必须只出现一次，不得因 failover 拼接导致两次")
+}
+
+// TestStreamWrittenGuard_GeminiPath_AbortFailoverOnSSEContentWritten 与上述测试相同，
+// 验证 Gemini 路径使用 service.PlatformGemini（而非 account.Platform）时行为一致。
+func TestStreamWrittenGuard_GeminiPath_AbortFailoverOnSSEContentWritten(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-2.0-flash:streamGenerateContent", nil)
+
+	sizeBeforeForward := c.Writer.Size()
+
+	_, err := c.Writer.Write([]byte(partialMessageStartSSE))
+	require.NoError(t, err)
+
+	require.NotEqual(t, sizeBeforeForward, c.Writer.Size())
+
+	failoverErr := &service.UpstreamFailoverError{
+		StatusCode: http.StatusForbidden,
+	}
+
+	h := &GatewayHandler{}
+	h.handleFailoverExhausted(c, failoverErr, service.PlatformGemini, true)
+
+	body := w.Body.String()
+
+	require.Contains(t, body, "event: message_start")
+	require.Contains(t, body, `"type":"error"`)
+
+	firstIdx := strings.Index(body, "event: message_start")
+	lastIdx := strings.LastIndex(body, "event: message_start")
+	assert.Equal(t, firstIdx, lastIdx, "Gemini 路径不得出现双 message_start")
+}
+
+// TestStreamWrittenGuard_NoByteWritten_GuardNotTriggered 验证反向场景：
+// 当 Forward 返回 UpstreamFailoverError 时若未向客户端写入任何 SSE 内容，
+// 守卫条件（c.Writer.Size() != sizeBeforeForward）为 false，不应中止 failover。
+func TestStreamWrittenGuard_NoByteWritten_GuardNotTriggered(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil)
+
+	// 模拟 writerSizeBeforeForward：初始为 -1
+	sizeBeforeForward := c.Writer.Size()
+
+	// Forward 未写入任何字节直接返回错误（例如 401 发生在连接建立前）
+	// c.Writer.Size() 仍为 -1
+
+	// 守卫条件：sizeBeforeForward == c.Writer.Size() → 不触发
+	guardTriggered := c.Writer.Size() != sizeBeforeForward
+	require.False(t, guardTriggered,
+		"未写入任何字节时，守卫条件必须为 false，应允许正常 failover 继续")
+}

backend/internal/handler/openai_chat_completions.go

@@ -256,14 +256,16 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
 
 		h.submitUsageRecordTask(func(ctx context.Context) {
 			if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
-				Result:        result,
-				APIKey:        apiKey,
-				User:          apiKey.User,
-				Account:       account,
-				Subscription:  subscription,
-				UserAgent:     userAgent,
-				IPAddress:     clientIP,
-				APIKeyService: h.apiKeyService,
+				Result:           result,
+				APIKey:           apiKey,
+				User:             apiKey.User,
+				Account:          account,
+				Subscription:     subscription,
+				InboundEndpoint:  normalizedOpenAIInboundEndpoint(c, openAIInboundEndpointChatCompletions),
+				UpstreamEndpoint: normalizedOpenAIUpstreamEndpoint(c, openAIUpstreamEndpointResponses),
+				UserAgent:        userAgent,
+				IPAddress:        clientIP,
+				APIKeyService:    h.apiKeyService,
 			}); err != nil {
 				logger.L().With(
 					zap.String("component", "handler.openai_gateway.chat_completions"),

backend/internal/handler/openai_gateway_endpoint_normalization_test.go

@@ -0,0 +1,57 @@
+package handler
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNormalizedOpenAIUpstreamEndpoint(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	tests := []struct {
+		name     string
+		path     string
+		fallback string
+		want     string
+	}{
+		{
+			name:     "responses root maps to responses upstream",
+			path:     "/v1/responses",
+			fallback: openAIUpstreamEndpointResponses,
+			want:     "/v1/responses",
+		},
+		{
+			name:     "responses compact keeps compact suffix",
+			path:     "/openai/v1/responses/compact",
+			fallback: openAIUpstreamEndpointResponses,
+			want:     "/v1/responses/compact",
+		},
+		{
+			name:     "responses nested suffix preserved",
+			path:     "/openai/v1/responses/compact/detail",
+			fallback: openAIUpstreamEndpointResponses,
+			want:     "/v1/responses/compact/detail",
+		},
+		{
+			name:     "non responses path uses fallback",
+			path:     "/v1/messages",
+			fallback: openAIUpstreamEndpointResponses,
+			want:     "/v1/responses",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rec := httptest.NewRecorder()
+			c, _ := gin.CreateTestContext(rec)
+			c.Request = httptest.NewRequest(http.MethodPost, tt.path, nil)
+
+			got := normalizedOpenAIUpstreamEndpoint(c, tt.fallback)
+			require.Equal(t, tt.want, got)
+		})
+	}
+}

backend/internal/handler/openai_gateway_handler.go

@@ -37,6 +37,13 @@ type OpenAIGatewayHandler struct {
 	cfg                     *config.Config
 }
 
+const (
+	openAIInboundEndpointResponses       = "/v1/responses"
+	openAIInboundEndpointMessages        = "/v1/messages"
+	openAIInboundEndpointChatCompletions = "/v1/chat/completions"
+	openAIUpstreamEndpointResponses      = "/v1/responses"
+)
+
 // NewOpenAIGatewayHandler creates a new OpenAIGatewayHandler
 func NewOpenAIGatewayHandler(
 	gatewayService *service.OpenAIGatewayService,
@@ -362,6 +369,8 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
 				User:               apiKey.User,
 				Account:            account,
 				Subscription:       subscription,
+				InboundEndpoint:    normalizedOpenAIInboundEndpoint(c, openAIInboundEndpointResponses),
+				UpstreamEndpoint:   normalizedOpenAIUpstreamEndpoint(c, openAIUpstreamEndpointResponses),
 				UserAgent:          userAgent,
 				IPAddress:          clientIP,
 				RequestPayloadHash: requestPayloadHash,
@@ -738,6 +747,8 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
 				User:               apiKey.User,
 				Account:            account,
 				Subscription:       subscription,
+				InboundEndpoint:    normalizedOpenAIInboundEndpoint(c, openAIInboundEndpointMessages),
+				UpstreamEndpoint:   normalizedOpenAIUpstreamEndpoint(c, openAIUpstreamEndpointResponses),
 				UserAgent:          userAgent,
 				IPAddress:          clientIP,
 				RequestPayloadHash: requestPayloadHash,
@@ -1235,6 +1246,8 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
 					User:               apiKey.User,
 					Account:            account,
 					Subscription:       subscription,
+					InboundEndpoint:    normalizedOpenAIInboundEndpoint(c, openAIInboundEndpointResponses),
+					UpstreamEndpoint:   normalizedOpenAIUpstreamEndpoint(c, openAIUpstreamEndpointResponses),
 					UserAgent:          userAgent,
 					IPAddress:          clientIP,
 					RequestPayloadHash: service.HashUsageRequestPayload(firstMessage),
@@ -1530,6 +1543,62 @@ func openAIWSIngressFallbackSessionSeed(userID, apiKeyID int64, groupID *int64)
 	return fmt.Sprintf("openai_ws_ingress:%d:%d:%d", gid, userID, apiKeyID)
 }
 
+func normalizedOpenAIInboundEndpoint(c *gin.Context, fallback string) string {
+	path := strings.TrimSpace(fallback)
+	if c != nil {
+		if fullPath := strings.TrimSpace(c.FullPath()); fullPath != "" {
+			path = fullPath
+		} else if c.Request != nil && c.Request.URL != nil {
+			if requestPath := strings.TrimSpace(c.Request.URL.Path); requestPath != "" {
+				path = requestPath
+			}
+		}
+	}
+
+	switch {
+	case strings.Contains(path, openAIInboundEndpointChatCompletions):
+		return openAIInboundEndpointChatCompletions
+	case strings.Contains(path, openAIInboundEndpointMessages):
+		return openAIInboundEndpointMessages
+	case strings.Contains(path, openAIInboundEndpointResponses):
+		return openAIInboundEndpointResponses
+	default:
+		return path
+	}
+}
+
+func normalizedOpenAIUpstreamEndpoint(c *gin.Context, fallback string) string {
+	base := strings.TrimSpace(fallback)
+	if base == "" {
+		base = openAIUpstreamEndpointResponses
+	}
+	base = strings.TrimRight(base, "/")
+
+	if c == nil || c.Request == nil || c.Request.URL == nil {
+		return base
+	}
+
+	path := strings.TrimRight(strings.TrimSpace(c.Request.URL.Path), "/")
+	if path == "" {
+		return base
+	}
+
+	idx := strings.LastIndex(path, "/responses")
+	if idx < 0 {
+		return base
+	}
+
+	suffix := strings.TrimSpace(path[idx+len("/responses"):])
+	if suffix == "" || suffix == "/" {
+		return base
+	}
+	if !strings.HasPrefix(suffix, "/") {
+		return base
+	}
+
+	return base + suffix
+}
+
 func isOpenAIWSUpgradeRequest(r *http.Request) bool {
 	if r == nil {
 		return false

backend/internal/handler/sora_gateway_handler_test.go

@@ -334,6 +334,14 @@ func (s *stubUsageLogRepo) GetUsageTrendWithFilters(ctx context.Context, startTi
 func (s *stubUsageLogRepo) GetModelStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) ([]usagestats.ModelStat, error) {
 	return nil, nil
 }
+
+func (s *stubUsageLogRepo) GetEndpointStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8) ([]usagestats.EndpointStat, error) {
+	return []usagestats.EndpointStat{}, nil
+}
+
+func (s *stubUsageLogRepo) GetUpstreamEndpointStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8) ([]usagestats.EndpointStat, error) {
+	return []usagestats.EndpointStat{}, nil
+}
 func (s *stubUsageLogRepo) GetGroupStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) ([]usagestats.GroupStat, error) {
 	return nil, nil
 }