fix(gemini): 原生路径登记 2xx 响应里的带内错误信号

Gemini API-key 原生转发对上游 2xx 只抽 usage 原样透传：finishReason 内容拦截、
promptFeedback.blockReason、事件里的 Google 错误信封、空响应都被当作成功，
ops 错误日志、用户错误页、渠道监控全部没有记录。

- 新增 gemini_response_signal：逐事件判定三类带内信号，流式按优先级登记最高者；
  没有 data 事件时对非 SSE 的兜底正文整段判定，空体登记 EMPTY_STREAM / EMPTY_RESPONSE
- 内容策略类按请求级结果登记（invalid_request_error / 400，business limited，
  不做上游归因、不计 SLA）；错误信封、生成异常、空响应按上游失败登记并计 SLA
- OpsStreamError 新增 RequestScoped / NonStream：请求级带内行不受残留上游上下文与
  skip_monitoring 影响，落库状态取 IntendedStatus；中间件保留此前尝试的恢复行
- 客户端字节、wire 状态、用量与计费不变

backend/internal/handler/ops_error_logger.go

@@ -1124,8 +1124,12 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc {
 				// A marked in-band error is a visible request failure even though its
 				// wire status is already 200. Otherwise retain recovered attempts as a
 				// provider-health row whose 2xx status keeps it outside request SLA.
-				if len(service.GetOpsStreamErrors(c)) > 0 {
+				if streamErrs := service.GetOpsStreamErrors(c); len(streamErrs) > 0 {
 					logOpsStreamError(c, ops, status)
+					// 请求级带内结果不承载上游归因，此前尝试的上游错误仍按恢复行记录。
+					if opsStreamErrorsAllRequestScoped(streamErrs) {
+						logOpsRecoveredUpstream(c, ops, status)
+					}
 				} else {
 					logOpsRecoveredUpstream(c, ops, status)
 				}
@@ -1405,22 +1409,31 @@ func opsRequestTypeFromContext(c *gin.Context) *int16 {
 	return nil
 }
 
-// logOpsStreamError 记录一次挂在已固化 HTTP 200 SSE 流上的就地错误。
-// 由于 wire 状态码停留在 200，常规的 status>=400 捕获路径永远不会触发；
-// handleStreamingAwareError 通过 service.MarkOpsStreamError 标记这类错误，
-// 此函数据此补记一条错误日志，让并发限流/流内失败在错误看板里可见。
-//
-// 仅在 status<400 且不存在上游错误上下文时调用：上游透传错误已由中间件的
-// upstream-context 分支落库，无需在此重复记录。
+// logOpsStreamError 记录挂在 2xx 响应上的带内错误（就地 SSE error 帧、非流式正文里的
+// 请求级结果等）。由于 wire 状态码停留在 2xx，常规的 status>=400 捕获路径不会触发；
+// 标记方通过 service.MarkOpsStreamError / MarkOpsStreamErrorValue 登记，此函数据此补记
+// 错误日志。上游错误上下文（若有）是否参与分类与归因由标记的 RequestScoped 决定。
 func logOpsStreamError(c *gin.Context, ops *service.OpsService, wireStatus int) {
 	for _, streamErr := range service.GetOpsStreamErrors(c) {
 		logOpsStreamErrorValue(c, ops, wireStatus, streamErr)
 	}
 }
 
+func opsStreamErrorsAllRequestScoped(streamErrs []service.OpsStreamError) bool {
+	if len(streamErrs) == 0 {
+		return false
+	}
+	for _, streamErr := range streamErrs {
+		if !streamErr.RequestScoped {
+			return false
+		}
+	}
+	return true
+}
+
 func logOpsStreamErrorValue(c *gin.Context, ops *service.OpsService, wireStatus int, streamErr service.OpsStreamError) {
 	// 命中 skip_monitoring=true 透传规则的请求跳过落库，与其它分支一致。
-	if streamErr.SkipMonitoring || (streamErr.Turn == 0 && shouldSkipFinalOpsFailure(c)) {
+	if streamErr.SkipMonitoring || (streamErr.Turn == 0 && !streamErr.RequestScoped && shouldSkipFinalOpsFailure(c)) {
 		return
 	}
 
@@ -1435,9 +1448,19 @@ func logOpsStreamErrorValue(c *gin.Context, ops *service.OpsService, wireStatus
 		classifyStatus = wireStatus
 	}
 	normalizedType := normalizeOpsErrorType(streamErr.ErrType, streamErr.Code)
-	phase, isBusinessLimited, errorOwner, errorSource := classifyOpsErrorLog(c, normalizedType, streamErr.Message, streamErr.Code, classifyStatus)
+	var phase, errorOwner, errorSource string
+	var isBusinessLimited bool
+	if streamErr.RequestScoped {
+		// 请求级带内结果只按错误类型分类，此前尝试残留的上游错误上下文不参与判定。
+		phase = classifyOpsPhase(normalizedType, streamErr.Message, streamErr.Code)
+		isBusinessLimited = true
+		errorOwner = classifyOpsErrorOwner(phase, streamErr.Message)
+		errorSource = classifyOpsErrorSource(phase, streamErr.Message)
+	} else {
+		phase, isBusinessLimited, errorOwner, errorSource = classifyOpsErrorLog(c, normalizedType, streamErr.Message, streamErr.Code, classifyStatus)
+	}
 	recordedStatus := wireStatus
-	if streamErr.CountTowardsSLA && streamErr.IntendedStatus >= 400 {
+	if streamErr.IntendedStatus >= 400 && (streamErr.CountTowardsSLA || streamErr.RequestScoped) {
 		recordedStatus = streamErr.IntendedStatus
 	}
 	errorBody := ""
@@ -1488,8 +1511,8 @@ func logOpsStreamErrorValue(c *gin.Context, ops *service.OpsService, wireStatus
 			}
 			return ""
 		}(),
-		// 就地 SSE 错误只出现在流式请求上。
-		Stream:           true,
+		// 带内错误默认挂在 SSE 流上；NonStream 标记的来自非流式 2xx 响应体。
+		Stream:           !streamErr.NonStream,
 		InboundEndpoint:  GetInboundEndpoint(c),
 		UpstreamEndpoint: GetUpstreamEndpoint(c, platform),
 		RequestedModel:   modelName,
@@ -1530,8 +1553,10 @@ func logOpsStreamErrorValue(c *gin.Context, ops *service.OpsService, wireStatus
 		CreatedAt: time.Now(),
 	}
 	applyOpsLatencyFieldsFromContext(c, entry)
-	applyOpsUpstreamFieldsFromContext(c, entry)
-	if streamErr.Turn > 0 {
+	if !streamErr.RequestScoped {
+		applyOpsUpstreamFieldsFromContext(c, entry)
+	}
+	if streamErr.Turn > 0 && !streamErr.RequestScoped {
 		applyOpsStreamErrorSnapshot(entry, streamErr)
 	}
 

backend/internal/handler/ops_error_logger_test.go

@@ -1996,3 +1996,175 @@ func TestClassifyOpsLocalModelConfigurationWithoutIngressMarkStaysRouting(t *tes
 	require.True(t, isBusinessLimited)
 	require.Equal(t, "platform", errorOwner)
 }
+
+// 带内错误也可能出现在非流式 2xx 响应体里（如 Gemini generateContent 的 finishReason）：
+// NonStream 标记落库 stream=false，未标记的沿用流式。
+func TestLogOpsStreamError_NonStreamInBandContentPolicy(t *testing.T) {
+	setupOpsErrorLogTestQueue(t, 4)
+
+	gin.SetMode(gin.TestMode)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-3.7-flash:generateContent", nil)
+	setOpsRequestContext(c, "gemini-3.7-flash", false)
+
+	service.MarkOpsStreamErrorValue(c, service.OpsStreamError{
+		ErrType:        "invalid_request_error",
+		Code:           "PROHIBITED_CONTENT",
+		Message:        "Gemini content policy stop (finishReason=PROHIBITED_CONTENT): Response stopped due to prohibited content.",
+		IntendedStatus: http.StatusBadRequest,
+		RequestScoped:  true,
+		NonStream:      true,
+	})
+
+	ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	logOpsStreamError(c, ops, http.StatusOK)
+
+	job := <-opsErrorLogQueue
+	require.NotNil(t, job.entry)
+	require.False(t, job.entry.Stream)
+	require.Equal(t, "invalid_request_error", job.entry.ErrorType)
+	require.Equal(t, "request", job.entry.ErrorPhase)
+	require.Equal(t, "client", job.entry.ErrorOwner)
+	require.True(t, job.entry.IsBusinessLimited)
+	require.Equal(t, http.StatusBadRequest, job.entry.StatusCode, "请求级带内结果按 IntendedStatus 落库，才能进入错误列表")
+	require.Equal(t, "P3", job.entry.Severity)
+	require.Equal(t, "gemini-3.7-flash", job.entry.Model)
+	require.Contains(t, job.entry.ErrorBody, "PROHIBITED_CONTENT")
+}
+
+// 请求级带内结果（RequestScoped）：此前尝试残留的上游错误上下文不参与分类、不做上游归因，
+// 按业务限制计；透传规则的 skip_monitoring 残留也不影响它落库。
+func TestLogOpsStreamError_RequestScopedIgnoresResidualUpstreamContext(t *testing.T) {
+	setupOpsErrorLogTestQueue(t, 4)
+
+	gin.SetMode(gin.TestMode)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-3.7-flash:streamGenerateContent", nil)
+	setOpsRequestContext(c, "gemini-3.7-flash", true)
+	c.Set(service.OpsUpstreamStatusCodeKey, http.StatusTooManyRequests)
+	c.Set(service.OpsUpstreamErrorMessageKey, "earlier attempt was rate limited")
+	c.Set(service.OpsUpstreamErrorsKey, []*service.OpsUpstreamErrorEvent{{
+		UpstreamStatusCode: http.StatusTooManyRequests,
+		Message:            "earlier attempt was rate limited",
+		SkipMonitoring:     true,
+	}})
+
+	service.MarkOpsStreamErrorValue(c, service.OpsStreamError{
+		ErrType:        "invalid_request_error",
+		Code:           "PROHIBITED_CONTENT",
+		Message:        "Gemini content policy stop (finishReason=PROHIBITED_CONTENT): Response stopped due to prohibited content.",
+		IntendedStatus: http.StatusBadRequest,
+		RequestScoped:  true,
+	})
+
+	ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	logOpsStreamError(c, ops, http.StatusOK)
+
+	require.Equal(t, int64(1), OpsErrorLogQueueLength())
+	job := <-opsErrorLogQueue
+	require.NotNil(t, job.entry)
+	require.Equal(t, "invalid_request_error", job.entry.ErrorType)
+	require.Equal(t, "request", job.entry.ErrorPhase)
+	require.Equal(t, "client", job.entry.ErrorOwner)
+	require.Equal(t, "client_request", job.entry.ErrorSource)
+	require.True(t, job.entry.IsBusinessLimited)
+	require.True(t, job.entry.Stream)
+	require.Equal(t, http.StatusBadRequest, job.entry.StatusCode)
+	require.Nil(t, job.entry.UpstreamStatusCode, "残留的 429 不得贴到内容策略行上")
+	require.Nil(t, job.entry.UpstreamErrorMessage)
+	require.Nil(t, job.entry.UpstreamErrorsJSON)
+	require.Contains(t, job.entry.ErrorBody, "PROHIBITED_CONTENT")
+}
+
+// 请求级带内结果与此前尝试的上游错误并存时，中间件既落带内错误行，也保留恢复行的 provider 遥测。
+func TestOpsErrorLoggerMiddleware_RequestScopedInBandErrorKeepsRecoveredTelemetry(t *testing.T) {
+	setupOpsErrorLogTestQueue(t, 4)
+	gin.SetMode(gin.TestMode)
+
+	ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	router := gin.New()
+	router.Use(OpsErrorLoggerMiddleware(ops))
+	router.POST("/v1beta/models/gemini-3.7-flash:generateContent", func(c *gin.Context) {
+		setOpsRequestContext(c, "gemini-3.7-flash", false)
+		c.Set(service.OpsUpstreamErrorsKey, []*service.OpsUpstreamErrorEvent{{
+			UpstreamStatusCode: http.StatusTooManyRequests,
+			Message:            "earlier attempt was rate limited",
+		}})
+		service.MarkOpsStreamErrorValue(c, service.OpsStreamError{
+			ErrType:        "invalid_request_error",
+			Code:           "SAFETY",
+			Message:        "Gemini content policy stop (finishReason=SAFETY): Response stopped due to safety reasons.",
+			IntendedStatus: http.StatusBadRequest,
+			RequestScoped:  true,
+			NonStream:      true,
+		})
+		c.JSON(http.StatusOK, gin.H{"candidates": []any{}})
+	})
+
+	recorder := httptest.NewRecorder()
+	router.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-3.7-flash:generateContent", nil))
+	require.Equal(t, http.StatusOK, recorder.Code)
+
+	require.Equal(t, int64(2), OpsErrorLogQueueLength())
+	inBand := <-opsErrorLogQueue
+	require.Equal(t, "invalid_request_error", inBand.entry.ErrorType)
+	require.Equal(t, "request", inBand.entry.ErrorPhase)
+	require.Equal(t, "client", inBand.entry.ErrorOwner)
+	require.True(t, inBand.entry.IsBusinessLimited)
+	require.False(t, inBand.entry.Stream)
+	require.Equal(t, http.StatusBadRequest, inBand.entry.StatusCode)
+	require.Nil(t, inBand.entry.UpstreamStatusCode)
+	require.Nil(t, inBand.entry.UpstreamErrorsJSON)
+
+	recovered := <-opsErrorLogQueue
+	require.Equal(t, "upstream", recovered.entry.ErrorPhase)
+	require.Equal(t, "provider", recovered.entry.ErrorOwner)
+	require.Equal(t, http.StatusOK, recovered.entry.StatusCode)
+	require.Equal(t, "Recovered upstream error 429: earlier attempt was rate limited", recovered.entry.ErrorMessage)
+}
+
+// 非请求级带内失败（如上游错误信封）仍与既有语义一致：上游上下文参与分类并做归因，只落一行。
+func TestOpsErrorLoggerMiddleware_UpstreamInBandFailureStillSingleRow(t *testing.T) {
+	setupOpsErrorLogTestQueue(t, 4)
+	gin.SetMode(gin.TestMode)
+
+	ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	router := gin.New()
+	router.Use(OpsErrorLoggerMiddleware(ops))
+	router.POST("/v1beta/models/gemini-3.7-flash:streamGenerateContent", func(c *gin.Context) {
+		setOpsRequestContext(c, "gemini-3.7-flash", true)
+		service.SetOpsUpstreamError(c, http.StatusTooManyRequests, "quota", "")
+		c.Set(service.OpsUpstreamErrorsKey, []*service.OpsUpstreamErrorEvent{{
+			UpstreamStatusCode: http.StatusTooManyRequests,
+			Message:            "quota",
+			Kind:               "stream_failed",
+		}})
+		service.MarkOpsStreamFailure(c, "rate_limit_error", "RESOURCE_EXHAUSTED", "quota", http.StatusTooManyRequests)
+		c.Data(http.StatusOK, "text/event-stream", []byte("data: {\"error\":{\"code\":429}}\n\n"))
+	})
+
+	recorder := httptest.NewRecorder()
+	router.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-3.7-flash:streamGenerateContent", nil))
+
+	require.Equal(t, int64(1), OpsErrorLogQueueLength())
+	job := <-opsErrorLogQueue
+	require.Equal(t, "rate_limit_error", job.entry.ErrorType)
+	require.Equal(t, "upstream", job.entry.ErrorPhase)
+	require.Equal(t, "provider", job.entry.ErrorOwner)
+	require.Equal(t, http.StatusTooManyRequests, job.entry.StatusCode)
+	require.True(t, job.entry.Stream)
+	require.NotNil(t, job.entry.UpstreamStatusCode)
+	require.Equal(t, http.StatusTooManyRequests, *job.entry.UpstreamStatusCode)
+}
+
+// Gemini 带内信号登记使用的错误类型必须都在 ops 白名单里，否则会被归一化成 api_error。
+func TestNormalizeOpsErrorType_KeepsGeminiInBandSignalTypes(t *testing.T) {
+	for _, errType := range []string{
+		"invalid_request_error", "authentication_error", "permission_error",
+		"not_found_error", "rate_limit_error", "upstream_error",
+	} {
+		require.Equal(t, errType, normalizeOpsErrorType(errType, "PROHIBITED_CONTENT"), errType)
+	}
+}

backend/internal/service/gemini_image_output_accounting_test.go

@@ -195,7 +195,7 @@ func TestHandleNativeNonStreamingResponse_FeedsImageCounter(t *testing.T) {
 	}
 
 	svc := &GeminiMessagesCompatService{}
-	usage, err := svc.handleNativeNonStreamingResponse(c, resp, false)
+	usage, err := svc.handleNativeNonStreamingResponse(c, resp, false, nil, "")
 	require.NoError(t, err)
 	require.NotNil(t, usage)
 

backend/internal/service/gemini_messages_compat_service.go

@@ -1591,7 +1591,7 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
 	var firstTokenMs *int
 
 	if stream {
-		streamRes, err := s.handleNativeStreamingResponse(c, resp, startTime, isOAuth)
+		streamRes, err := s.handleNativeStreamingResponse(c, resp, startTime, isOAuth, account, requestID)
 		if err != nil {
 			return nil, err
 		}
@@ -1599,17 +1599,23 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
 		firstTokenMs = streamRes.firstTokenMs
 	} else {
 		if useUpstreamStream {
-			collected, usageObj, err := collectGeminiSSE(resp.Body, isOAuth)
+			var best geminiResponseSignal
+			collected, usageObj, stats, err := collectGeminiSSEObserved(resp.Body, isOAuth, func(rawBytes []byte) {
+				if sig, ok := detectGeminiResponseSignal(rawBytes); ok && sig.Kind > best.Kind {
+					best = sig
+				}
+			})
 			if err != nil {
 				return nil, s.writeGoogleError(c, http.StatusBadGateway, "Failed to read upstream stream")
 			}
+			s.finalizeGeminiSSESignal(c, account, false, requestID, best, stats.dataEvents > 0, stats.fallback)
 			b, _ := json.Marshal(collected)
 			upstreamResponseModelObserverFromContext(c).ObserveGemini(b)
 			observeGeminiImageOutputs(c, b)
 			c.Data(http.StatusOK, "application/json", b)
 			usage = usageObj
 		} else {
-			usageResp, err := s.handleNativeNonStreamingResponse(c, resp, isOAuth)
+			usageResp, err := s.handleNativeNonStreamingResponse(c, resp, isOAuth, account, requestID)
 			if err != nil {
 				return nil, err
 			}
@@ -2438,23 +2444,40 @@ func unwrapIfNeeded(isOAuth bool, raw []byte) []byte {
 }
 
 func collectGeminiSSE(body io.Reader, isOAuth bool) (map[string]any, *ClaudeUsage, error) {
+	collected, usage, _, err := collectGeminiSSEObserved(body, isOAuth, nil)
+	return collected, usage, err
+}
+
+// geminiSSECollectStats 记录一次 SSE 聚合读到的 data 事件数，以及非 data 行的兜底内容。
+type geminiSSECollectStats struct {
+	dataEvents int
+	fallback   *geminiSSEFallbackBody
+}
+
+// collectGeminiSSEObserved 在聚合的同时把每个解包后的事件原文交给 observe（可为 nil）。
+func collectGeminiSSEObserved(body io.Reader, isOAuth bool, observe func(rawBytes []byte)) (map[string]any, *ClaudeUsage, geminiSSECollectStats, error) {
 	reader := bufio.NewReader(body)
 
 	var last map[string]any
 	var lastWithParts map[string]any
 	var collectedTextParts []string // Collect all text parts for aggregation
 	usage := &ClaudeUsage{}
+	stats := geminiSSECollectStats{fallback: &geminiSSEFallbackBody{}}
 
 	for {
 		line, err := reader.ReadString('\n')
 		if len(line) > 0 {
 			trimmed := strings.TrimRight(line, "\r\n")
-			if strings.HasPrefix(trimmed, "data:") {
+			if !strings.HasPrefix(trimmed, "data:") {
+				if stats.dataEvents == 0 {
+					stats.fallback.AddLine(trimmed)
+				}
+			} else {
 				payload := strings.TrimSpace(strings.TrimPrefix(trimmed, "data:"))
 				switch payload {
 				case "", "[DONE]":
 					if payload == "[DONE]" {
-						return mergeCollectedTextParts(pickGeminiCollectResult(last, lastWithParts), collectedTextParts), usage, nil
+						return mergeCollectedTextParts(pickGeminiCollectResult(last, lastWithParts), collectedTextParts), usage, stats, nil
 					}
 				default:
 					var parsed map[string]any
@@ -2469,6 +2492,12 @@ func collectGeminiSSE(body io.Reader, isOAuth bool) (map[string]any, *ClaudeUsag
 						rawBytes = []byte(payload)
 						_ = json.Unmarshal(rawBytes, &parsed)
 					}
+					if len(rawBytes) > 0 {
+						stats.dataEvents++
+						if observe != nil {
+							observe(rawBytes)
+						}
+					}
 					if parsed != nil {
 						last = parsed
 						if u := extractGeminiUsage(rawBytes); u != nil {
@@ -2492,11 +2521,11 @@ func collectGeminiSSE(body io.Reader, isOAuth bool) (map[string]any, *ClaudeUsag
 			break
 		}
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, stats, err
 		}
 	}
 
-	return mergeCollectedTextParts(pickGeminiCollectResult(last, lastWithParts), collectedTextParts), usage, nil
+	return mergeCollectedTextParts(pickGeminiCollectResult(last, lastWithParts), collectedTextParts), usage, stats, nil
 }
 
 func pickGeminiCollectResult(last map[string]any, lastWithParts map[string]any) map[string]any {
@@ -2657,7 +2686,7 @@ type UpstreamHTTPResult struct {
 	Body       []byte
 }
 
-func (s *GeminiMessagesCompatService) handleNativeNonStreamingResponse(c *gin.Context, resp *http.Response, isOAuth bool) (*ClaudeUsage, error) {
+func (s *GeminiMessagesCompatService) handleNativeNonStreamingResponse(c *gin.Context, resp *http.Response, isOAuth bool, account *Account, upstreamRequestID string) (*ClaudeUsage, error) {
 	if s.cfg != nil && s.cfg.Gateway.GeminiDebugResponseHeaders {
 		logger.LegacyPrintf("service.gemini_messages_compat", "[GeminiAPI] ========== Response Headers ==========")
 		for key, values := range resp.Header {
@@ -2685,6 +2714,11 @@ func (s *GeminiMessagesCompatService) handleNativeNonStreamingResponse(c *gin.Co
 	}
 	observer.ObserveGemini(respBody)
 	observeGeminiImageOutputs(c, respBody)
+	if sig, ok := detectGeminiResponseSignalInBody(respBody); ok {
+		s.markGeminiResponseSignal(c, account, sig, false, upstreamRequestID)
+	} else if isGeminiEmptyResponseBody(respBody) {
+		s.markGeminiEmptyResponse(c, account, false, upstreamRequestID)
+	}
 
 	responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
 
@@ -2700,7 +2734,7 @@ func (s *GeminiMessagesCompatService) handleNativeNonStreamingResponse(c *gin.Co
 	return &ClaudeUsage{}, nil
 }
 
-func (s *GeminiMessagesCompatService) handleNativeStreamingResponse(c *gin.Context, resp *http.Response, startTime time.Time, isOAuth bool) (*geminiNativeStreamResult, error) {
+func (s *GeminiMessagesCompatService) handleNativeStreamingResponse(c *gin.Context, resp *http.Response, startTime time.Time, isOAuth bool, account *Account, upstreamRequestID string) (*geminiNativeStreamResult, error) {
 	if s.cfg != nil && s.cfg.Gateway.GeminiDebugResponseHeaders {
 		logger.LegacyPrintf("service.gemini_messages_compat", "[GeminiAPI] ========== Streaming Response Headers ==========")
 		for key, values := range resp.Header {
@@ -2738,6 +2772,9 @@ func (s *GeminiMessagesCompatService) handleNativeStreamingResponse(c *gin.Conte
 		observer = beginUpstreamResponseModelObservation(c)
 	}
 	var firstTokenMs *int
+	var best geminiResponseSignal
+	sawDataEvent := false
+	fallback := &geminiSSEFallbackBody{}
 
 	for {
 		line, err := reader.ReadString('\n')
@@ -2764,6 +2801,10 @@ func (s *GeminiMessagesCompatService) handleNativeStreamingResponse(c *gin.Conte
 						rawBytes = []byte(payload)
 					}
 
+					sawDataEvent = true
+					if sig, ok := detectGeminiResponseSignal(rawBytes); ok && sig.Kind > best.Kind {
+						best = sig
+					}
 					if u := extractGeminiUsage(rawBytes); u != nil {
 						usage = u
 					}
@@ -2785,6 +2826,9 @@ func (s *GeminiMessagesCompatService) handleNativeStreamingResponse(c *gin.Conte
 					flusher.Flush()
 				}
 			} else {
+				if !sawDataEvent {
+					fallback.AddLine(trimmed)
+				}
 				_, _ = io.WriteString(c.Writer, line)
 				flusher.Flush()
 			}
@@ -2798,6 +2842,8 @@ func (s *GeminiMessagesCompatService) handleNativeStreamingResponse(c *gin.Conte
 		}
 	}
 
+	s.finalizeGeminiSSESignal(c, account, true, upstreamRequestID, best, sawDataEvent, fallback)
+
 	return &geminiNativeStreamResult{usage: usage, firstTokenMs: firstTokenMs}, nil
 }
 

backend/internal/service/gemini_messages_compat_service_test.go

@@ -564,7 +564,7 @@ func TestGeminiHandleNativeNonStreamingResponse_DebugDisabledDoesNotEmitHeaderLo
 		Body: io.NopCloser(strings.NewReader(`{"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":2}}`)),
 	}
 
-	usage, err := svc.handleNativeNonStreamingResponse(c, resp, false)
+	usage, err := svc.handleNativeNonStreamingResponse(c, resp, false, nil, "")
 	require.NoError(t, err)
 	require.NotNil(t, usage)
 	require.False(t, logSink.ContainsMessage("[GeminiAPI]"), "debug 关闭时不应输出 Gemini 响应头日志")

backend/internal/service/gemini_native_response_signal_test.go

@@ -0,0 +1,487 @@
+//go:build unit
+
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/config"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+// ---------------------------------------------------------------------------
+// Gemini 原生路径带内信号契约：
+//   - 上游 2xx 里的 finishReason / promptFeedback.blockReason / 错误信封 / 空流只登记 ops，
+//     wire 状态与客户端字节原样不变，用量照常从 usageMetadata 提取（计费不受影响）；
+//   - 内容策略类不归因上游账号、不计入 SLA；错误信封与空流按上游失败归因并计入 SLA。
+// ---------------------------------------------------------------------------
+
+const geminiSignalTestFinishMessage = "The model output could not be generated. This output contains sensitive words that violate Google's [Generative AI Prohibited Use policy](https://policies.google.com/terms/generative-ai/use-policy). If you think this was an error, [send feedback](https://ai.google.dev/gemini-api/docs/troubleshooting)."
+
+const geminiSignalTestProhibitedSSE = `data: {"candidates":[{"content":{"parts":[{"text":"hello"}],"role":"model"}}],"modelVersion":"gemini-3.7-flash","responseId":"r1","usageMetadata":{"candidatesTokenCount":26,"promptTokenCount":21775,"thoughtsTokenCount":606,"totalTokenCount":22407}}
+
+data: {"candidates":[{"content":{"parts":[{"text":" world"}],"role":"model"}}],"modelVersion":"gemini-3.7-flash","responseId":"r1","usageMetadata":{"candidatesTokenCount":1289,"promptTokenCount":21775,"thoughtsTokenCount":606,"totalTokenCount":23670}}
+
+data: {"candidates":[{"content":{},"finishMessage":"` + geminiSignalTestFinishMessage + `","finishReason":"PROHIBITED_CONTENT"}],"modelVersion":"gemini-3.7-flash","responseId":"r1","usageMetadata":{"candidatesTokenCount":1289,"promptTokenCount":21775,"thoughtsTokenCount":606,"totalTokenCount":23670}}
+
+`
+
+const geminiSignalTestStopSSE = `data: {"candidates":[{"content":{"parts":[{"text":"hello"}],"role":"model"}}],"usageMetadata":{"candidatesTokenCount":3,"promptTokenCount":10,"totalTokenCount":13}}
+
+data: {"candidates":[{"content":{"parts":[{"text":"!"}],"role":"model"},"finishReason":"STOP"}],"usageMetadata":{"candidatesTokenCount":4,"promptTokenCount":10,"totalTokenCount":14}}
+
+`
+
+func geminiSignalTestAccount() *Account {
+	return &Account{
+		ID:       703,
+		Name:     "gemini-signal-test",
+		Platform: PlatformGemini,
+		Type:     AccountTypeAPIKey,
+		Credentials: map[string]any{
+			"api_key": "test-key",
+		},
+	}
+}
+
+func newGeminiSignalService(contentType string, body string) *GeminiMessagesCompatService {
+	httpStub := &geminiCompatHTTPUpstreamStub{
+		response: &http.Response{
+			StatusCode: http.StatusOK,
+			Header:     http.Header{"Content-Type": []string{contentType}, "X-Request-Id": []string{"upstream-req-1"}},
+			Body:       io.NopCloser(strings.NewReader(body)),
+		},
+	}
+	return &GeminiMessagesCompatService{
+		httpUpstream:     httpStub,
+		cfg:              &config.Config{},
+		rateLimitService: NewRateLimitService(&errorPolicyRepoStub{}, nil, &config.Config{}, nil, nil),
+	}
+}
+
+func geminiSignalTestRequest() []byte {
+	return []byte(`{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}`)
+}
+
+func upstreamErrorEventsFromContext(t *testing.T, c *gin.Context) []*OpsUpstreamErrorEvent {
+	t.Helper()
+	v, ok := c.Get(OpsUpstreamErrorsKey)
+	if !ok {
+		return nil
+	}
+	events, ok := v.([]*OpsUpstreamErrorEvent)
+	require.True(t, ok)
+	return events
+}
+
+func TestGeminiForwardNative_StreamProhibitedContentMarksInBandErrorAndKeepsBillingUsage(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	svc := newGeminiSignalService("text/event-stream; charset=utf-8", geminiSignalTestProhibitedSSE)
+	c, rec := newGeminiNativeTestContext(t)
+
+	result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	// 客户端字节原样透传，wire 状态 200。
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.Equal(t, geminiSignalTestProhibitedSSE, rec.Body.String())
+
+	// 用量照常提取：input=promptTokenCount，output=candidates+thoughts。
+	require.Equal(t, 21775, result.Usage.InputTokens)
+	require.Equal(t, 1289+606, result.Usage.OutputTokens)
+	require.NotNil(t, result.FirstTokenMs)
+
+	// 带内错误登记为请求级内容策略：不计 SLA、不归因上游账号。
+	streamErrs := GetOpsStreamErrors(c)
+	require.Len(t, streamErrs, 1)
+	require.Equal(t, "invalid_request_error", streamErrs[0].ErrType)
+	require.Equal(t, "PROHIBITED_CONTENT", streamErrs[0].Code)
+	require.Equal(t, http.StatusBadRequest, streamErrs[0].IntendedStatus)
+	require.False(t, streamErrs[0].CountTowardsSLA)
+	require.True(t, streamErrs[0].RequestScoped)
+	require.False(t, streamErrs[0].NonStream)
+	require.Contains(t, streamErrs[0].Message, "finishReason=PROHIBITED_CONTENT")
+	require.Contains(t, streamErrs[0].Message, "Prohibited Use policy")
+
+	_, hasUpstreamStatus := c.Get(OpsUpstreamStatusCodeKey)
+	require.False(t, hasUpstreamStatus, "内容策略不应写上游错误上下文")
+	require.Empty(t, upstreamErrorEventsFromContext(t, c))
+}
+
+func TestGeminiForwardNative_StreamErrorEnvelopeMarksUpstreamFailure(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	body := `data: {"candidates":[{"content":{"parts":[{"text":"partial"}],"role":"model"}}],"usageMetadata":{"candidatesTokenCount":2,"promptTokenCount":10,"totalTokenCount":12}}
+
+data: {"error":{"code":429,"message":"Resource has been exhausted (e.g. check quota).","status":"RESOURCE_EXHAUSTED"}}
+
+`
+	svc := newGeminiSignalService("text/event-stream", body)
+	c, rec := newGeminiNativeTestContext(t)
+
+	result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Equal(t, body, rec.Body.String(), "错误信封也原样透传，交给 SDK 客户端按错误处理")
+	require.Equal(t, 10, result.Usage.InputTokens)
+	require.Equal(t, 2, result.Usage.OutputTokens)
+
+	streamErrs := GetOpsStreamErrors(c)
+	require.Len(t, streamErrs, 1)
+	require.Equal(t, "rate_limit_error", streamErrs[0].ErrType)
+	require.Equal(t, "RESOURCE_EXHAUSTED", streamErrs[0].Code)
+	require.Equal(t, http.StatusTooManyRequests, streamErrs[0].IntendedStatus)
+	require.True(t, streamErrs[0].CountTowardsSLA)
+	require.False(t, streamErrs[0].RequestScoped)
+	require.False(t, streamErrs[0].NonStream)
+	require.Equal(t, "Resource has been exhausted (e.g. check quota).", streamErrs[0].Message)
+
+	status, ok := c.Get(OpsUpstreamStatusCodeKey)
+	require.True(t, ok)
+	require.Equal(t, http.StatusTooManyRequests, status)
+
+	events := upstreamErrorEventsFromContext(t, c)
+	require.Len(t, events, 1)
+	require.Equal(t, "stream_failed", events[0].Kind)
+	require.Equal(t, int64(703), events[0].AccountID)
+	require.Equal(t, PlatformGemini, events[0].Platform)
+	require.Equal(t, http.StatusTooManyRequests, events[0].UpstreamStatusCode)
+	require.Equal(t, "upstream-req-1", events[0].UpstreamRequestID)
+}
+
+func TestGeminiForwardNative_StreamPromptBlockedMarksContentPolicy(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	body := `data: {"promptFeedback":{"blockReason":"SAFETY","safetyRatings":[{"category":"HARM_CATEGORY_HARASSMENT","probability":"HIGH","blocked":true}]},"usageMetadata":{"promptTokenCount":10,"totalTokenCount":10}}
+
+`
+	svc := newGeminiSignalService("text/event-stream", body)
+	c, rec := newGeminiNativeTestContext(t)
+
+	result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.Equal(t, body, rec.Body.String())
+	require.Equal(t, 10, result.Usage.InputTokens)
+
+	streamErrs := GetOpsStreamErrors(c)
+	require.Len(t, streamErrs, 1)
+	require.Equal(t, "invalid_request_error", streamErrs[0].ErrType)
+	require.Equal(t, "SAFETY", streamErrs[0].Code)
+	require.Contains(t, streamErrs[0].Message, "blockReason=SAFETY")
+	require.False(t, streamErrs[0].CountTowardsSLA)
+	require.Empty(t, upstreamErrorEventsFromContext(t, c))
+}
+
+func TestGeminiForwardNative_EmptyStreamMarksUpstreamFailure(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	svc := newGeminiSignalService("text/event-stream", "")
+	c, rec := newGeminiNativeTestContext(t)
+
+	result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.Zero(t, rec.Body.Len())
+	require.Nil(t, result.FirstTokenMs)
+
+	streamErrs := GetOpsStreamErrors(c)
+	require.Len(t, streamErrs, 1)
+	require.Equal(t, "upstream_error", streamErrs[0].ErrType)
+	require.Equal(t, geminiSignalEmptyStreamReason, streamErrs[0].Code)
+	require.Equal(t, http.StatusBadGateway, streamErrs[0].IntendedStatus)
+	require.True(t, streamErrs[0].CountTowardsSLA)
+
+	events := upstreamErrorEventsFromContext(t, c)
+	require.Len(t, events, 1)
+	require.Equal(t, "stream_failed", events[0].Kind)
+	require.Equal(t, int64(703), events[0].AccountID)
+}
+
+func TestGeminiForwardNative_NormalStreamLeavesNoMark(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	svc := newGeminiSignalService("text/event-stream", geminiSignalTestStopSSE)
+	c, rec := newGeminiNativeTestContext(t)
+
+	result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.Equal(t, geminiSignalTestStopSSE, rec.Body.String())
+	require.Equal(t, 4, result.Usage.OutputTokens)
+	require.Empty(t, GetOpsStreamErrors(c))
+	require.Empty(t, upstreamErrorEventsFromContext(t, c))
+	_, hasUpstreamStatus := c.Get(OpsUpstreamStatusCodeKey)
+	require.False(t, hasUpstreamStatus)
+}
+
+func TestGeminiForwardNative_NonStreamProhibitedContentMarksInBandError(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	body := `{"candidates":[{"content":{},"finishMessage":"` + geminiSignalTestFinishMessage + `","finishReason":"PROHIBITED_CONTENT"}],"modelVersion":"gemini-3.7-flash","usageMetadata":{"candidatesTokenCount":120,"promptTokenCount":900,"thoughtsTokenCount":30,"totalTokenCount":1050}}`
+	svc := newGeminiSignalService("application/json", body)
+	c, rec := newGeminiNativeTestContext(t)
+
+	result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "generateContent", false, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.Equal(t, body, rec.Body.String())
+	require.Equal(t, 900, result.Usage.InputTokens)
+	require.Equal(t, 150, result.Usage.OutputTokens)
+
+	streamErrs := GetOpsStreamErrors(c)
+	require.Len(t, streamErrs, 1)
+	require.Equal(t, "invalid_request_error", streamErrs[0].ErrType)
+	require.Equal(t, "PROHIBITED_CONTENT", streamErrs[0].Code)
+	require.False(t, streamErrs[0].CountTowardsSLA)
+	require.True(t, streamErrs[0].RequestScoped)
+	require.True(t, streamErrs[0].NonStream)
+	require.Empty(t, upstreamErrorEventsFromContext(t, c))
+}
+
+func TestGeminiForwardNative_NonStreamErrorEnvelopeOn200MarksUpstreamFailure(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	body := `{"error":{"code":500,"message":"Internal error encountered.","status":"INTERNAL"}}`
+	svc := newGeminiSignalService("application/json", body)
+	c, rec := newGeminiNativeTestContext(t)
+
+	result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "generateContent", false, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Equal(t, body, rec.Body.String())
+
+	streamErrs := GetOpsStreamErrors(c)
+	require.Len(t, streamErrs, 1)
+	require.Equal(t, "upstream_error", streamErrs[0].ErrType)
+	require.Equal(t, "INTERNAL", streamErrs[0].Code)
+	require.Equal(t, http.StatusInternalServerError, streamErrs[0].IntendedStatus)
+	require.True(t, streamErrs[0].CountTowardsSLA)
+
+	require.True(t, streamErrs[0].NonStream)
+
+	events := upstreamErrorEventsFromContext(t, c)
+	require.Len(t, events, 1)
+	require.Equal(t, "http_error", events[0].Kind)
+}
+
+func TestGeminiForwardNative_StreamErrorEnvelopeAfterContentFilterWins(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	body := `data: {"candidates":[{"content":{},"finishReason":"SAFETY"}]}
+
+data: {"error":{"code":503,"message":"The model is overloaded. Please try again later.","status":"UNAVAILABLE"}}
+
+`
+	svc := newGeminiSignalService("text/event-stream", body)
+	c, rec := newGeminiNativeTestContext(t)
+
+	_, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.Equal(t, body, rec.Body.String())
+
+	streamErrs := GetOpsStreamErrors(c)
+	require.Len(t, streamErrs, 1)
+	require.Equal(t, "UNAVAILABLE", streamErrs[0].Code, "错误信封优先于内容过滤")
+	require.Equal(t, "upstream_error", streamErrs[0].ErrType)
+	require.Equal(t, http.StatusServiceUnavailable, streamErrs[0].IntendedStatus)
+	require.True(t, streamErrs[0].CountTowardsSLA)
+	require.Len(t, upstreamErrorEventsFromContext(t, c), 1)
+}
+
+func TestGeminiForwardNative_StreamNonSSEJSONBodyIsInspected(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	t.Run("json array with content filter", func(t *testing.T) {
+		body := `[{
+  "candidates": [{"content": {"parts": [{"text": "hello"}], "role": "model"}}],
+  "usageMetadata": {"promptTokenCount": 10, "candidatesTokenCount": 1}
+}
+,
+{
+  "candidates": [{"content": {}, "finishReason": "PROHIBITED_CONTENT"}]
+}
+]
+`
+		svc := newGeminiSignalService("application/json", body)
+		c, rec := newGeminiNativeTestContext(t)
+		_, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+			"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+		require.NoError(t, err)
+		require.Equal(t, body, rec.Body.String())
+		streamErrs := GetOpsStreamErrors(c)
+		require.Len(t, streamErrs, 1)
+		require.Equal(t, "PROHIBITED_CONTENT", streamErrs[0].Code)
+		require.True(t, streamErrs[0].RequestScoped)
+	})
+	t.Run("json object error envelope", func(t *testing.T) {
+		body := `{"error":{"code":429,"message":"quota","status":"RESOURCE_EXHAUSTED"}}` + "\n"
+		svc := newGeminiSignalService("application/json", body)
+		c, _ := newGeminiNativeTestContext(t)
+		_, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+			"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+		require.NoError(t, err)
+		streamErrs := GetOpsStreamErrors(c)
+		require.Len(t, streamErrs, 1)
+		require.Equal(t, "RESOURCE_EXHAUSTED", streamErrs[0].Code)
+		require.True(t, streamErrs[0].CountTowardsSLA)
+	})
+	t.Run("healthy json object leaves no mark", func(t *testing.T) {
+		body := `{"candidates":[{"content":{"parts":[{"text":"hello"}],"role":"model"},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":1}}` + "\n"
+		svc := newGeminiSignalService("application/json", body)
+		c, _ := newGeminiNativeTestContext(t)
+		_, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+			"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+		require.NoError(t, err)
+		require.Empty(t, GetOpsStreamErrors(c))
+		require.Empty(t, upstreamErrorEventsFromContext(t, c))
+	})
+}
+
+func TestGeminiForwardNative_StreamOversizedNonSSEBodyLeavesNoMark(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	filler := strings.Repeat("x", geminiSSEFallbackBodyLimit)
+	body := `{"candidates":[{"content":{"parts":[{"text":"` + filler + `"}],"role":"model"},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":1}}` + "\n"
+	svc := newGeminiSignalService("application/json", body)
+	c, rec := newGeminiNativeTestContext(t)
+	_, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.Equal(t, len(body), rec.Body.Len())
+	require.Empty(t, GetOpsStreamErrors(c), "超限的兜底体放弃判定，不得记成空流")
+	require.Empty(t, upstreamErrorEventsFromContext(t, c))
+}
+
+func TestGeminiForwardNative_StreamNonSSEEmptyBodyIsEmpty(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for name, body := range map[string]string{
+		"empty object":     "{}\n",
+		"empty candidates": `{"candidates":[],"usageMetadata":{"promptTokenCount":3}}` + "\n",
+		"empty array":      "[]\n",
+		"array of empties": "[{\"candidates\":[]}\n,\n{}]\n",
+	} {
+		t.Run(name, func(t *testing.T) {
+			svc := newGeminiSignalService("application/json", body)
+			c, rec := newGeminiNativeTestContext(t)
+			_, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+				"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+			require.NoError(t, err)
+			require.Equal(t, body, rec.Body.String())
+			streamErrs := GetOpsStreamErrors(c)
+			require.Len(t, streamErrs, 1)
+			require.Equal(t, geminiSignalEmptyStreamReason, streamErrs[0].Code)
+			require.True(t, streamErrs[0].CountTowardsSLA)
+		})
+	}
+}
+
+func TestGeminiForwardNative_StreamWithoutDataEventsIsEmpty(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for name, body := range map[string]string{
+		"keepalive comments only": ": keepalive\n\n: keepalive\n\n",
+		"done marker only":        "data: [DONE]\n\n",
+		"blank data only":         "data: \n\n",
+		"whitespace only":         "\n\n   \n",
+	} {
+		t.Run(name, func(t *testing.T) {
+			svc := newGeminiSignalService("text/event-stream", body)
+			c, rec := newGeminiNativeTestContext(t)
+			_, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+				"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+			require.NoError(t, err)
+			require.Equal(t, body, rec.Body.String())
+			streamErrs := GetOpsStreamErrors(c)
+			require.Len(t, streamErrs, 1)
+			require.Equal(t, geminiSignalEmptyStreamReason, streamErrs[0].Code)
+			require.True(t, streamErrs[0].CountTowardsSLA)
+			require.False(t, streamErrs[0].NonStream)
+		})
+	}
+}
+
+func TestGeminiForwardNative_StreamOtherFinishReasonLeavesNoMark(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	body := `data: {"candidates":[{"content":{"parts":[{"text":"full answer"}],"role":"model"},"finishReason":"OTHER"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":2}}
+
+`
+	svc := newGeminiSignalService("text/event-stream", body)
+	c, _ := newGeminiNativeTestContext(t)
+	result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "streamGenerateContent", true, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.Equal(t, 2, result.Usage.OutputTokens)
+	require.Empty(t, GetOpsStreamErrors(c))
+	require.Empty(t, upstreamErrorEventsFromContext(t, c))
+}
+
+func TestGeminiForwardNative_NonStreamEmptyBodyMarksEmptyResponse(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for name, body := range map[string]string{
+		"blank":            "",
+		"empty object":     "{}",
+		"empty candidates": `{"candidates":[],"usageMetadata":{"promptTokenCount":3,"totalTokenCount":3}}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			svc := newGeminiSignalService("application/json", body)
+			c, rec := newGeminiNativeTestContext(t)
+			result, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+				"gemini-3.7-flash", "generateContent", false, geminiSignalTestRequest())
+			require.NoError(t, err)
+			require.NotNil(t, result)
+			require.Equal(t, body, rec.Body.String())
+			streamErrs := GetOpsStreamErrors(c)
+			require.Len(t, streamErrs, 1)
+			require.Equal(t, geminiSignalEmptyResponseReason, streamErrs[0].Code)
+			require.Equal(t, "upstream_error", streamErrs[0].ErrType)
+			require.True(t, streamErrs[0].CountTowardsSLA)
+			require.True(t, streamErrs[0].NonStream)
+			events := upstreamErrorEventsFromContext(t, c)
+			require.Len(t, events, 1)
+			require.Equal(t, "http_error", events[0].Kind)
+		})
+	}
+}
+
+func TestGeminiForwardNative_CountTokensBodyLeavesNoMark(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	svc := newGeminiSignalService("application/json", `{"totalTokens":12}`)
+	c, rec := newGeminiNativeTestContext(t)
+	_, err := svc.ForwardNative(context.Background(), c, geminiSignalTestAccount(),
+		"gemini-3.7-flash", "countTokens", false, geminiSignalTestRequest())
+	require.NoError(t, err)
+	require.Equal(t, `{"totalTokens":12}`, rec.Body.String())
+	require.Empty(t, GetOpsStreamErrors(c))
+	require.Empty(t, upstreamErrorEventsFromContext(t, c))
+}
+
+func TestCollectGeminiSSEObserved_SeesTerminalChunkEvenWhenAggregateDropsIt(t *testing.T) {
+	var observed []string
+	collected, usage, stats, err := collectGeminiSSEObserved(strings.NewReader(geminiSignalTestProhibitedSSE), false, func(raw []byte) {
+		observed = append(observed, string(raw))
+	})
+	require.NoError(t, err)
+	require.Len(t, observed, 3)
+	require.Equal(t, 3, stats.dataEvents)
+	require.Empty(t, stats.fallback.Bytes())
+	require.Equal(t, 21775, usage.InputTokens)
+
+	var hit bool
+	for _, raw := range observed {
+		if sig, ok := detectGeminiResponseSignal([]byte(raw)); ok {
+			hit = true
+			require.Equal(t, geminiSignalContentFilter, sig.Kind)
+		}
+	}
+	require.True(t, hit)
+	// 聚合结果沿用"最后一个带 parts 的块"，本身不携带末块的 finishReason；检测必须逐事件进行。
+	aggregated, err := json.Marshal(collected)
+	require.NoError(t, err)
+	require.NotContains(t, string(aggregated), "PROHIBITED_CONTENT")
+}

backend/internal/service/gemini_response_signal.go

@@ -0,0 +1,446 @@
+package service
+
+import (
+	"bytes"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/tidwall/gjson"
+)
+
+// geminiResponseSignalKind 区分 Gemini 2xx 响应事件里携带的带内信号。
+// 这些信号都承载在 HTTP 2xx 之上，客户端字节原样透传，只影响 ops 记录。
+// 数值越大优先级越高：同一响应出现多个信号时只登记优先级最高的一个。
+type geminiResponseSignalKind int
+
+const (
+	geminiSignalNone geminiResponseSignalKind = iota
+	// geminiSignalPromptBlocked：promptFeedback.blockReason 有值，提示词整体被拦，candidates 为空。
+	geminiSignalPromptBlocked
+	// geminiSignalContentFilter：首候选 finishReason 属于内容过滤类枚举。
+	geminiSignalContentFilter
+	// geminiSignalAbnormalStop：首候选 finishReason 是模型生成异常，或响应体为空。
+	geminiSignalAbnormalStop
+	// geminiSignalError：事件本身是 Google 错误信封 {"error":{"code","message","status"}}，
+	// 与 Gemini SDK 客户端对流式 chunk 的错误判定相同。
+	geminiSignalError
+)
+
+// geminiResponseSignal 是一次带内信号的归一化描述。
+type geminiResponseSignal struct {
+	Kind geminiResponseSignalKind
+	// Reason 是 finishReason / blockReason 枚举值，或 Google 错误信封的 status；作为 ops 错误 code。
+	Reason  string
+	Message string
+	// Status 是该信号的语义 HTTP 状态：内容策略类固定 400，错误信封取 error.code，生成异常与空响应 502。
+	Status int
+	// Detail 是错误信封原文（已截断），仅 geminiSignalError 填写。
+	Detail string
+}
+
+const (
+	geminiSignalEmptyStreamReason   = "EMPTY_STREAM"
+	geminiSignalEmptyResponseReason = "EMPTY_RESPONSE"
+	// geminiSSEFallbackBodyLimit 是非 data 行兜底内容的收集上限，超过即放弃整段判定。
+	geminiSSEFallbackBodyLimit = 1 << 20
+)
+
+// geminiSignalProbeKeys 是三类信号各自必然出现的字段名；事件不含任何一个即可跳过 JSON 解析。
+var geminiSignalProbeKeys = [][]byte{[]byte(`"error"`), []byte(`"promptFeedback"`), []byte(`"finishReason"`)}
+
+func geminiPayloadMayCarrySignal(payload []byte) bool {
+	for _, key := range geminiSignalProbeKeys {
+		if bytes.Contains(payload, key) {
+			return true
+		}
+	}
+	return false
+}
+
+// detectGeminiResponseSignal 检查一个 Gemini 响应事件是否携带带内信号；Code Assist 的
+// {"response":{...}} 包装体先解包。判定顺序：错误信封 > promptFeedback.blockReason > 首候选 finishReason。
+// finishReason 为空、STOP、MAX_TOKENS、FINISH_REASON_UNSPECIFIED 视为正常；OTHER、
+// TOO_MANY_TOOL_CALLS、NO_IMAGE、IMAGE_OTHER 与未知枚举同样按正常停止处理（与 gemini-cli 遥测一致）。
+func detectGeminiResponseSignal(payload []byte) (geminiResponseSignal, bool) {
+	payload = bytes.TrimSpace(payload)
+	if len(payload) == 0 || !geminiPayloadMayCarrySignal(payload) || !gjson.ValidBytes(payload) {
+		return geminiResponseSignal{}, false
+	}
+	if inner := gjson.GetBytes(payload, "response"); inner.Exists() && inner.IsObject() {
+		payload = []byte(inner.Raw)
+	}
+
+	if errObj := gjson.GetBytes(payload, "error"); errObj.Exists() && errObj.IsObject() {
+		googleStatus := strings.ToUpper(strings.TrimSpace(errObj.Get("status").String()))
+		status := int(errObj.Get("code").Int())
+		if status < 400 || status > 599 {
+			status = geminiGoogleStatusToHTTPStatus(googleStatus)
+		}
+		message := strings.TrimSpace(errObj.Get("message").String())
+		if message == "" {
+			message = "Gemini upstream returned an error event"
+		}
+		reason := googleStatus
+		if reason == "" {
+			reason = "UPSTREAM_ERROR"
+		}
+		return geminiResponseSignal{
+			Kind:    geminiSignalError,
+			Reason:  reason,
+			Message: message,
+			Status:  status,
+			Detail:  truncateString(string(payload), 2048),
+		}, true
+	}
+
+	blockReason := strings.ToUpper(strings.TrimSpace(gjson.GetBytes(payload, "promptFeedback.blockReason").String()))
+	switch blockReason {
+	case "", "BLOCKED_REASON_UNSPECIFIED":
+	default:
+		return geminiResponseSignal{
+			Kind:    geminiSignalPromptBlocked,
+			Reason:  blockReason,
+			Message: fmt.Sprintf("Gemini content policy block (blockReason=%s): prompt was blocked before generation", blockReason),
+			Status:  http.StatusBadRequest,
+		}, true
+	}
+
+	candidate := geminiPrimaryCandidate(payload)
+	finishReason := strings.ToUpper(strings.TrimSpace(candidate.Get("finishReason").String()))
+	switch finishReason {
+	case "", "STOP", "MAX_TOKENS", "FINISH_REASON_UNSPECIFIED":
+		return geminiResponseSignal{}, false
+	}
+	detail := strings.TrimSpace(candidate.Get("finishMessage").String())
+	if detail == "" {
+		detail = geminiFinishReasonText(finishReason)
+	}
+	detail = truncateString(detail, 512)
+	if isGeminiContentFilterFinishReason(finishReason) {
+		return geminiResponseSignal{
+			Kind:    geminiSignalContentFilter,
+			Reason:  finishReason,
+			Message: fmt.Sprintf("Gemini content policy stop (finishReason=%s): %s", finishReason, detail),
+			Status:  http.StatusBadRequest,
+		}, true
+	}
+	if isGeminiAbnormalStopFinishReason(finishReason) {
+		return geminiResponseSignal{
+			Kind:    geminiSignalAbnormalStop,
+			Reason:  finishReason,
+			Message: fmt.Sprintf("Gemini generation stopped abnormally (finishReason=%s): %s", finishReason, detail),
+			Status:  http.StatusBadGateway,
+		}, true
+	}
+	return geminiResponseSignal{}, false
+}
+
+// geminiPrimaryCandidate 返回首候选：index 缺省或为 0 的候选。多候选流里其它 index 的候选不参与判定。
+func geminiPrimaryCandidate(payload []byte) gjson.Result {
+	candidates := gjson.GetBytes(payload, "candidates")
+	if !candidates.IsArray() {
+		return gjson.Result{}
+	}
+	var primary gjson.Result
+	candidates.ForEach(func(_, cand gjson.Result) bool {
+		if index := cand.Get("index"); index.Exists() && index.Int() != 0 {
+			return true
+		}
+		primary = cand
+		return false
+	})
+	return primary
+}
+
+// detectGeminiResponseSignalInBody 对整段响应体判定：JSON 数组逐元素判定并取优先级最高者，对象直接判定。
+func detectGeminiResponseSignalInBody(body []byte) (geminiResponseSignal, bool) {
+	body = bytes.TrimSpace(body)
+	if len(body) == 0 || !geminiPayloadMayCarrySignal(body) || !gjson.ValidBytes(body) {
+		return geminiResponseSignal{}, false
+	}
+	parsed := gjson.ParseBytes(body)
+	if !parsed.IsArray() {
+		return detectGeminiResponseSignal(body)
+	}
+	var best geminiResponseSignal
+	parsed.ForEach(func(_, item gjson.Result) bool {
+		if sig, ok := detectGeminiResponseSignal([]byte(item.Raw)); ok && sig.Kind > best.Kind {
+			best = sig
+		}
+		return true
+	})
+	return best, best.Kind != geminiSignalNone
+}
+
+// isGeminiEmptyResponseBody 判定 2xx 响应体是否没有任何响应内容：空白、空对象、candidates 为空数组
+// 且没有 promptFeedback，或数组为空 / 全部元素都满足前述条件；Code Assist 的 response 包装先解包。
+func isGeminiEmptyResponseBody(body []byte) bool {
+	body = bytes.TrimSpace(body)
+	if len(body) == 0 {
+		return true
+	}
+	if !gjson.ValidBytes(body) {
+		return false
+	}
+	parsed := gjson.ParseBytes(body)
+	if parsed.IsArray() {
+		empty := true
+		parsed.ForEach(func(_, item gjson.Result) bool {
+			empty = isGeminiEmptyResponseObject(item)
+			return empty
+		})
+		return empty
+	}
+	return isGeminiEmptyResponseObject(parsed)
+}
+
+func isGeminiEmptyResponseObject(item gjson.Result) bool {
+	if !item.IsObject() {
+		return false
+	}
+	if inner := item.Get("response"); inner.Exists() && inner.IsObject() {
+		item = inner
+	}
+	empty := true
+	item.ForEach(func(_, _ gjson.Result) bool {
+		empty = false
+		return false
+	})
+	if empty {
+		return true
+	}
+	candidates := item.Get("candidates")
+	return candidates.IsArray() && len(candidates.Array()) == 0 && !item.Get("promptFeedback").Exists()
+}
+
+// isGeminiContentFilterFinishReason 的集合与 gemini-cli 遥测的 CONTENT_FILTER 归类一致，并纳入同语义的图片枚举。
+func isGeminiContentFilterFinishReason(reason string) bool {
+	switch reason {
+	case "SAFETY", "RECITATION", "LANGUAGE", "BLOCKLIST", "PROHIBITED_CONTENT", "SPII",
+		"IMAGE_SAFETY", "IMAGE_PROHIBITED_CONTENT", "IMAGE_RECITATION":
+		return true
+	default:
+		return false
+	}
+}
+
+// isGeminiAbnormalStopFinishReason 的集合与 gemini-cli 遥测的 ERROR 归类一致。
+func isGeminiAbnormalStopFinishReason(reason string) bool {
+	switch reason {
+	case "MALFORMED_FUNCTION_CALL", "UNEXPECTED_TOOL_CALL":
+		return true
+	default:
+		return false
+	}
+}
+
+// geminiFinishReasonText 是 finishMessage 缺失时的兜底描述；列出的枚举沿用 gemini-cli 的
+// 用户提示文案，其余枚举用通用兜底。
+func geminiFinishReasonText(reason string) string {
+	switch reason {
+	case "SAFETY":
+		return "Response stopped due to safety reasons."
+	case "RECITATION":
+		return "Response stopped due to recitation policy."
+	case "LANGUAGE":
+		return "Response stopped due to unsupported language."
+	case "BLOCKLIST":
+		return "Response stopped due to forbidden terms."
+	case "PROHIBITED_CONTENT":
+		return "Response stopped due to prohibited content."
+	case "SPII":
+		return "Response stopped due to sensitive personally identifiable information."
+	case "MALFORMED_FUNCTION_CALL":
+		return "Response stopped due to malformed function call."
+	case "IMAGE_SAFETY":
+		return "Response stopped due to image safety violations."
+	case "UNEXPECTED_TOOL_CALL":
+		return "Response stopped due to unexpected tool call."
+	case "IMAGE_PROHIBITED_CONTENT":
+		return "Response stopped due to prohibited image content."
+	default:
+		return "Response stopped."
+	}
+}
+
+// geminiGoogleStatusToHTTPStatus 把 google.rpc.Code 名称映射为 HTTP 状态；未知取 502。
+func geminiGoogleStatusToHTTPStatus(status string) int {
+	switch status {
+	case "INVALID_ARGUMENT", "FAILED_PRECONDITION", "OUT_OF_RANGE":
+		return http.StatusBadRequest
+	case "UNAUTHENTICATED":
+		return http.StatusUnauthorized
+	case "PERMISSION_DENIED":
+		return http.StatusForbidden
+	case "NOT_FOUND":
+		return http.StatusNotFound
+	case "RESOURCE_EXHAUSTED":
+		return http.StatusTooManyRequests
+	case "INTERNAL", "UNKNOWN", "DATA_LOSS":
+		return http.StatusInternalServerError
+	case "UNAVAILABLE":
+		return http.StatusServiceUnavailable
+	case "DEADLINE_EXCEEDED":
+		return http.StatusGatewayTimeout
+	default:
+		return http.StatusBadGateway
+	}
+}
+
+// geminiSignalOpsErrorType 把语义状态映射为 ops 已知的错误类型（见 handler.isKnownOpsErrorType）。
+func geminiSignalOpsErrorType(status int) string {
+	switch status {
+	case http.StatusBadRequest:
+		return "invalid_request_error"
+	case http.StatusUnauthorized:
+		return "authentication_error"
+	case http.StatusForbidden:
+		return "permission_error"
+	case http.StatusNotFound:
+		return "not_found_error"
+	case http.StatusTooManyRequests:
+		return "rate_limit_error"
+	default:
+		return "upstream_error"
+	}
+}
+
+// geminiSSEFallbackBody 收集 SSE 读取过程中既非 data 行也非注释的非空内容（上游忽略 alt=sse
+// 直接回 JSON 的形态），超过上限后停止收集并标记截断。
+type geminiSSEFallbackBody struct {
+	buf       bytes.Buffer
+	truncated bool
+}
+
+func (b *geminiSSEFallbackBody) AddLine(line string) {
+	if b == nil || b.truncated {
+		return
+	}
+	trimmed := strings.TrimSpace(line)
+	if trimmed == "" || strings.HasPrefix(trimmed, ":") {
+		return
+	}
+	if b.buf.Len()+len(trimmed)+1 > geminiSSEFallbackBodyLimit {
+		b.truncated = true
+		return
+	}
+	_, _ = b.buf.WriteString(trimmed)
+	_ = b.buf.WriteByte('\n')
+}
+
+func (b *geminiSSEFallbackBody) Bytes() []byte {
+	if b == nil {
+		return nil
+	}
+	return b.buf.Bytes()
+}
+
+func (b *geminiSSEFallbackBody) Truncated() bool {
+	return b != nil && b.truncated
+}
+
+// markGeminiResponseSignal 把带内信号登记到 ops 上下文。wire 状态保持 2xx，客户端字节不变，
+// 用量与计费照常走 ForwardResult。stream 是客户端请求的流式标记。
+//   - 内容策略类（promptFeedback.blockReason / 内容过滤 finishReason）是请求级结果：
+//     只记请求级带内错误，不归因上游账号，不计入 SLA。
+//   - 错误信封、生成异常与空响应按上游失败登记：写上游错误上下文与尝试事件，并按语义状态计入 SLA。
+func (s *GeminiMessagesCompatService) markGeminiResponseSignal(c *gin.Context, account *Account, sig geminiResponseSignal, stream bool, upstreamRequestID string) {
+	if c == nil || sig.Kind == geminiSignalNone {
+		return
+	}
+	message := sanitizeUpstreamErrorMessage(strings.TrimSpace(sig.Message))
+	switch sig.Kind {
+	case geminiSignalPromptBlocked, geminiSignalContentFilter:
+		MarkOpsStreamErrorValue(c, OpsStreamError{
+			ErrType:        "invalid_request_error",
+			Code:           sig.Reason,
+			Message:        message,
+			IntendedStatus: sig.Status,
+			RequestScoped:  true,
+			NonStream:      !stream,
+		})
+		return
+	}
+
+	detail := ""
+	if sig.Detail != "" && s != nil && s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
+		maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
+		if maxBytes <= 0 {
+			maxBytes = 2048
+		}
+		detail = truncateString(sig.Detail, maxBytes)
+	}
+	setOpsUpstreamError(c, sig.Status, message, detail)
+	kind := "http_error"
+	if stream {
+		kind = "stream_failed"
+	}
+	event := OpsUpstreamErrorEvent{
+		ProxyID:            opsUpstreamProxyID(account),
+		ProxyName:          opsUpstreamProxyName(account),
+		Platform:           PlatformGemini,
+		UpstreamStatusCode: sig.Status,
+		UpstreamRequestID:  strings.TrimSpace(upstreamRequestID),
+		Kind:               kind,
+		Message:            message,
+		Detail:             detail,
+	}
+	if account != nil {
+		event.Platform = account.Platform
+		event.AccountID = account.ID
+		event.AccountName = account.Name
+	}
+	appendOpsUpstreamError(c, event)
+	MarkOpsStreamErrorValue(c, OpsStreamError{
+		ErrType:         geminiSignalOpsErrorType(sig.Status),
+		Code:            sig.Reason,
+		Message:         message,
+		IntendedStatus:  sig.Status,
+		CountTowardsSLA: true,
+		NonStream:       !stream,
+	})
+}
+
+// markGeminiEmptyResponse 登记"上游 2xx 但没有任何响应内容"的结果：流式指整段 body 没有 data 事件
+// 也没有其它非空内容，非流式指空体、空对象或 candidates 为空数组。只记录不换号。
+func (s *GeminiMessagesCompatService) markGeminiEmptyResponse(c *gin.Context, account *Account, stream bool, upstreamRequestID string) {
+	reason := geminiSignalEmptyResponseReason
+	message := "Gemini upstream returned an empty response body"
+	if stream {
+		reason = geminiSignalEmptyStreamReason
+		message = "Gemini upstream returned an empty stream body"
+	}
+	s.markGeminiResponseSignal(c, account, geminiResponseSignal{
+		Kind:    geminiSignalAbnormalStop,
+		Reason:  reason,
+		Message: message,
+		Status:  http.StatusBadGateway,
+	}, stream, upstreamRequestID)
+}
+
+// finalizeGeminiSSESignal 在上游 SSE 读完后做最终登记：优先登记流中优先级最高的信号；
+// 没有任何 data 事件时，兜底内容超限则放弃判定，为空登记空响应，是非 SSE 的 JSON 则对整段
+// 判定一次，整段没有信号但没有任何响应内容同样登记空响应。
+func (s *GeminiMessagesCompatService) finalizeGeminiSSESignal(c *gin.Context, account *Account, stream bool, upstreamRequestID string, best geminiResponseSignal, sawDataEvent bool, fallback *geminiSSEFallbackBody) {
+	if best.Kind != geminiSignalNone {
+		s.markGeminiResponseSignal(c, account, best, stream, upstreamRequestID)
+		return
+	}
+	if sawDataEvent || fallback.Truncated() {
+		return
+	}
+	body := fallback.Bytes()
+	if len(bytes.TrimSpace(body)) == 0 {
+		s.markGeminiEmptyResponse(c, account, stream, upstreamRequestID)
+		return
+	}
+	if sig, ok := detectGeminiResponseSignalInBody(body); ok {
+		s.markGeminiResponseSignal(c, account, sig, stream, upstreamRequestID)
+		return
+	}
+	if isGeminiEmptyResponseBody(body) {
+		s.markGeminiEmptyResponse(c, account, stream, upstreamRequestID)
+	}
+}

backend/internal/service/gemini_response_signal_test.go

@@ -0,0 +1,235 @@
+//go:build unit
+
+package service
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDetectGeminiResponseSignal(t *testing.T) {
+	prohibited := `{"candidates":[{"content":{},"finishMessage":"The model output could not be generated. This output contains sensitive words that violate Google's [Generative AI Prohibited Use policy](https://policies.google.com/terms/generative-ai/use-policy).","finishReason":"PROHIBITED_CONTENT"}],"usageMetadata":{"candidatesTokenCount":1289,"promptTokenCount":21775}}`
+
+	tests := []struct {
+		name        string
+		payload     string
+		wantHit     bool
+		wantKind    geminiResponseSignalKind
+		wantReason  string
+		wantStatus  int
+		wantMessage string
+	}{
+		{name: "empty", payload: "", wantHit: false},
+		{name: "invalid json", payload: "data: {", wantHit: false},
+		{name: "streaming chunk without finishReason", payload: `{"candidates":[{"content":{"parts":[{"text":"hi"}],"role":"model"}}]}`, wantHit: false},
+		{name: "text mentioning error is not an envelope", payload: `{"candidates":[{"content":{"parts":[{"text":"an \"error\" happened in the story"}]}}]}`, wantHit: false},
+		{name: "STOP", payload: `{"candidates":[{"content":{"parts":[{"text":"hi"}]},"finishReason":"STOP"}]}`, wantHit: false},
+		{name: "MAX_TOKENS", payload: `{"candidates":[{"content":{"parts":[{"text":"hi"}]},"finishReason":"MAX_TOKENS"}]}`, wantHit: false},
+		{name: "FINISH_REASON_UNSPECIFIED", payload: `{"candidates":[{"finishReason":"FINISH_REASON_UNSPECIFIED"}]}`, wantHit: false},
+		{name: "OTHER is a normal stop", payload: `{"candidates":[{"content":{"parts":[{"text":"full answer"}]},"finishReason":"OTHER"}]}`, wantHit: false},
+		{name: "TOO_MANY_TOOL_CALLS is a normal stop", payload: `{"candidates":[{"finishReason":"TOO_MANY_TOOL_CALLS"}]}`, wantHit: false},
+		{name: "NO_IMAGE is a normal stop", payload: `{"candidates":[{"finishReason":"NO_IMAGE"}]}`, wantHit: false},
+		{name: "unknown future finishReason is a normal stop", payload: `{"candidates":[{"finishReason":"SOME_NEW_REASON"}]}`, wantHit: false},
+		{name: "countTokens body", payload: `{"totalTokens":12}`, wantHit: false},
+		{name: "BLOCKED_REASON_UNSPECIFIED is not a block", payload: `{"promptFeedback":{"blockReason":"BLOCKED_REASON_UNSPECIFIED"}}`, wantHit: false},
+		{name: "secondary candidate is ignored", payload: `{"candidates":[{"index":1,"finishReason":"SAFETY"}]}`, wantHit: false},
+		{name: "error as string is not an envelope", payload: `{"error":"plain"}`, wantHit: false},
+		{name: "error as number is not an envelope", payload: `{"error":123}`, wantHit: false},
+		{
+			name: "code assist wrapped event is unwrapped", payload: `{"response":{"candidates":[{"content":{},"finishReason":"SAFETY"}]},"traceId":"t"}`, wantHit: true,
+			wantKind: geminiSignalContentFilter, wantReason: "SAFETY", wantStatus: http.StatusBadRequest,
+		},
+		{
+			name: "PROHIBITED_CONTENT with finishMessage", payload: prohibited, wantHit: true,
+			wantKind: geminiSignalContentFilter, wantReason: "PROHIBITED_CONTENT", wantStatus: http.StatusBadRequest,
+			wantMessage: "Gemini content policy stop (finishReason=PROHIBITED_CONTENT): The model output could not be generated.",
+		},
+		{
+			name: "SAFETY without finishMessage uses fallback text", payload: `{"candidates":[{"content":{},"finishReason":"SAFETY"}]}`, wantHit: true,
+			wantKind: geminiSignalContentFilter, wantReason: "SAFETY", wantStatus: http.StatusBadRequest,
+			wantMessage: "Gemini content policy stop (finishReason=SAFETY): Response stopped due to safety reasons.",
+		},
+		{
+			name: "lowercase finishReason is normalized", payload: `{"candidates":[{"finishReason":"blocklist"}]}`, wantHit: true,
+			wantKind: geminiSignalContentFilter, wantReason: "BLOCKLIST", wantStatus: http.StatusBadRequest,
+		},
+		{
+			name: "explicit index 0 candidate is primary", payload: `{"candidates":[{"index":1,"finishReason":"STOP"},{"index":0,"finishReason":"SPII"}]}`, wantHit: true,
+			wantKind: geminiSignalContentFilter, wantReason: "SPII", wantStatus: http.StatusBadRequest,
+		},
+		{
+			name: "MALFORMED_FUNCTION_CALL is an abnormal stop", payload: `{"candidates":[{"content":{},"finishReason":"MALFORMED_FUNCTION_CALL"}]}`, wantHit: true,
+			wantKind: geminiSignalAbnormalStop, wantReason: "MALFORMED_FUNCTION_CALL", wantStatus: http.StatusBadGateway,
+			wantMessage: "Gemini generation stopped abnormally (finishReason=MALFORMED_FUNCTION_CALL): Response stopped due to malformed function call.",
+		},
+		{
+			name: "UNEXPECTED_TOOL_CALL is an abnormal stop", payload: `{"candidates":[{"finishReason":"UNEXPECTED_TOOL_CALL"}]}`, wantHit: true,
+			wantKind: geminiSignalAbnormalStop, wantReason: "UNEXPECTED_TOOL_CALL", wantStatus: http.StatusBadGateway,
+		},
+		{
+			name: "promptFeedback blockReason", payload: `{"promptFeedback":{"blockReason":"PROHIBITED_CONTENT","safetyRatings":[]},"usageMetadata":{"promptTokenCount":10,"totalTokenCount":10}}`, wantHit: true,
+			wantKind: geminiSignalPromptBlocked, wantReason: "PROHIBITED_CONTENT", wantStatus: http.StatusBadRequest,
+			wantMessage: "Gemini content policy block (blockReason=PROHIBITED_CONTENT): prompt was blocked before generation",
+		},
+		{
+			name: "google error envelope with numeric code", payload: `{"error":{"code":429,"message":"Resource has been exhausted (e.g. check quota).","status":"RESOURCE_EXHAUSTED"}}`, wantHit: true,
+			wantKind: geminiSignalError, wantReason: "RESOURCE_EXHAUSTED", wantStatus: http.StatusTooManyRequests,
+			wantMessage: "Resource has been exhausted (e.g. check quota).",
+		},
+		{
+			name: "google error envelope without code maps status name", payload: `{"error":{"message":"Internal error encountered.","status":"INTERNAL"}}`, wantHit: true,
+			wantKind: geminiSignalError, wantReason: "INTERNAL", wantStatus: http.StatusInternalServerError,
+			wantMessage: "Internal error encountered.",
+		},
+		{
+			name: "error envelope with out-of-range code maps status name", payload: `{"error":{"code":200,"message":"weird","status":"resource_exhausted"}}`, wantHit: true,
+			wantKind: geminiSignalError, wantReason: "RESOURCE_EXHAUSTED", wantStatus: http.StatusTooManyRequests,
+		},
+		{
+			name: "error envelope without status or code falls back to 502", payload: `{"error":{"message":"boom"}}`, wantHit: true,
+			wantKind: geminiSignalError, wantReason: "UPSTREAM_ERROR", wantStatus: http.StatusBadGateway, wantMessage: "boom",
+		},
+		{
+			name: "error takes precedence over finishReason", payload: `{"error":{"code":503,"status":"UNAVAILABLE","message":"busy"},"candidates":[{"finishReason":"SAFETY"}]}`, wantHit: true,
+			wantKind: geminiSignalError, wantReason: "UNAVAILABLE", wantStatus: http.StatusServiceUnavailable,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sig, hit := detectGeminiResponseSignal([]byte(tt.payload))
+			require.Equal(t, tt.wantHit, hit)
+			if !tt.wantHit {
+				require.Equal(t, geminiSignalNone, sig.Kind)
+				return
+			}
+			require.Equal(t, tt.wantKind, sig.Kind)
+			require.Equal(t, tt.wantReason, sig.Reason)
+			require.Equal(t, tt.wantStatus, sig.Status)
+			if tt.wantMessage != "" {
+				require.Contains(t, sig.Message, tt.wantMessage)
+			}
+			if tt.wantKind == geminiSignalError {
+				require.NotEmpty(t, sig.Detail)
+			} else {
+				require.Empty(t, sig.Detail)
+			}
+		})
+	}
+}
+
+func TestGeminiResponseSignalKindPrecedence(t *testing.T) {
+	require.Greater(t, geminiSignalError, geminiSignalAbnormalStop)
+	require.Greater(t, geminiSignalAbnormalStop, geminiSignalContentFilter)
+	require.Greater(t, geminiSignalContentFilter, geminiSignalPromptBlocked)
+	require.Greater(t, geminiSignalPromptBlocked, geminiSignalNone)
+}
+
+func TestDetectGeminiResponseSignalInBody(t *testing.T) {
+	t.Run("json array picks the most severe element", func(t *testing.T) {
+		body := `[{"candidates":[{"content":{"parts":[{"text":"a"}]}}]}
+,
+{"candidates":[{"content":{},"finishReason":"SAFETY"}]}
+,
+{"error":{"code":503,"status":"UNAVAILABLE","message":"busy"}}
+]`
+		sig, ok := detectGeminiResponseSignalInBody([]byte(body))
+		require.True(t, ok)
+		require.Equal(t, geminiSignalError, sig.Kind)
+		require.Equal(t, "UNAVAILABLE", sig.Reason)
+	})
+	t.Run("json array without signal", func(t *testing.T) {
+		_, ok := detectGeminiResponseSignalInBody([]byte(`[{"candidates":[{"content":{"parts":[{"text":"a"}]},"finishReason":"STOP"}]}]`))
+		require.False(t, ok)
+	})
+	t.Run("code assist wrapped object is unwrapped", func(t *testing.T) {
+		sig, ok := detectGeminiResponseSignalInBody([]byte(`{"response":{"candidates":[{"content":{},"finishReason":"PROHIBITED_CONTENT"}]},"traceId":"t"}`))
+		require.True(t, ok)
+		require.Equal(t, geminiSignalContentFilter, sig.Kind)
+		require.Equal(t, "PROHIBITED_CONTENT", sig.Reason)
+	})
+	t.Run("plain object", func(t *testing.T) {
+		sig, ok := detectGeminiResponseSignalInBody([]byte(`{"promptFeedback":{"blockReason":"SAFETY"}}`))
+		require.True(t, ok)
+		require.Equal(t, geminiSignalPromptBlocked, sig.Kind)
+	})
+	t.Run("non json", func(t *testing.T) {
+		_, ok := detectGeminiResponseSignalInBody([]byte(`<html>error</html>`))
+		require.False(t, ok)
+	})
+}
+
+func TestIsGeminiEmptyResponseBody(t *testing.T) {
+	require.True(t, isGeminiEmptyResponseBody(nil))
+	require.True(t, isGeminiEmptyResponseBody([]byte("  \n")))
+	require.True(t, isGeminiEmptyResponseBody([]byte(`{}`)))
+	require.True(t, isGeminiEmptyResponseBody([]byte(`{"candidates":[]}`)))
+	require.True(t, isGeminiEmptyResponseBody([]byte(`{"candidates":[],"usageMetadata":{"promptTokenCount":3}}`)))
+	require.False(t, isGeminiEmptyResponseBody([]byte(`{"candidates":[],"promptFeedback":{"safetyRatings":[]}}`)))
+	require.False(t, isGeminiEmptyResponseBody([]byte(`{"totalTokens":12}`)))
+	require.False(t, isGeminiEmptyResponseBody([]byte(`{"candidates":[{"content":{"parts":[{"text":"hi"}]}}]}`)))
+	require.True(t, isGeminiEmptyResponseBody([]byte(`[]`)))
+	require.True(t, isGeminiEmptyResponseBody([]byte(`[{}]`)))
+	require.True(t, isGeminiEmptyResponseBody([]byte(`[{"candidates":[]},{}]`)))
+	require.False(t, isGeminiEmptyResponseBody([]byte(`[{"candidates":[]},{"candidates":[{"content":{}}]}]`)))
+	require.False(t, isGeminiEmptyResponseBody([]byte(`[1]`)))
+	require.True(t, isGeminiEmptyResponseBody([]byte(`{"response":{}}`)))
+	require.True(t, isGeminiEmptyResponseBody([]byte(`{"response":{"candidates":[]},"traceId":"t"}`)))
+	require.False(t, isGeminiEmptyResponseBody([]byte(`{"response":{"totalTokens":3}}`)))
+	require.False(t, isGeminiEmptyResponseBody([]byte(`not json`)))
+}
+
+func TestMarkOpsStreamErrorValue_RequestScopedSkipsUpstreamSnapshot(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	newCtx := func() *gin.Context {
+		c, _ := gin.CreateTestContext(httptest.NewRecorder())
+		c.Request = httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-3.7-flash:generateContent", nil)
+		setOpsUpstreamError(c, http.StatusTooManyRequests, "earlier attempt was rate limited", "detail")
+		appendOpsUpstreamError(c, OpsUpstreamErrorEvent{UpstreamStatusCode: http.StatusTooManyRequests, Message: "earlier attempt was rate limited", SkipMonitoring: true})
+		return c
+	}
+
+	scoped := newCtx()
+	MarkOpsStreamErrorValue(scoped, OpsStreamError{ErrType: "invalid_request_error", Code: "SAFETY", Message: "m", IntendedStatus: http.StatusBadRequest, RequestScoped: true})
+	got, ok := GetOpsStreamError(scoped)
+	require.True(t, ok)
+	require.Zero(t, got.UpstreamStatus)
+	require.Empty(t, got.UpstreamMessage)
+	require.Empty(t, got.UpstreamDetail)
+	require.Nil(t, got.UpstreamErrors)
+	require.False(t, got.SkipMonitoring)
+
+	plain := newCtx()
+	MarkOpsStreamErrorValue(plain, OpsStreamError{ErrType: "rate_limit_error", Code: "RESOURCE_EXHAUSTED", Message: "m", IntendedStatus: http.StatusTooManyRequests, CountTowardsSLA: true})
+	got, ok = GetOpsStreamError(plain)
+	require.True(t, ok)
+	require.Equal(t, http.StatusTooManyRequests, got.UpstreamStatus)
+	require.Equal(t, "earlier attempt was rate limited", got.UpstreamMessage)
+	require.Len(t, got.UpstreamErrors, 1)
+	require.True(t, got.SkipMonitoring)
+}
+
+func TestGeminiSSEFallbackBody(t *testing.T) {
+	var fb geminiSSEFallbackBody
+	fb.AddLine("")
+	fb.AddLine("   ")
+	fb.AddLine(": keepalive")
+	require.Empty(t, fb.Bytes())
+	fb.AddLine(`{"a":1}`)
+	fb.AddLine("event: ping")
+	require.Equal(t, "{\"a\":1}\nevent: ping\n", string(fb.Bytes()))
+	require.False(t, fb.Truncated())
+
+	var big geminiSSEFallbackBody
+	big.AddLine(string(make([]byte, geminiSSEFallbackBodyLimit)))
+	require.True(t, big.Truncated())
+	require.Empty(t, big.Bytes())
+	var nilBody *geminiSSEFallbackBody
+	nilBody.AddLine("x")
+	require.Nil(t, nilBody.Bytes())
+	require.False(t, nilBody.Truncated())
+}

backend/internal/service/ops_upstream_context.go

@@ -134,12 +134,11 @@ func OpsClientBusinessLimitedReason(c *gin.Context) string {
 	return strings.TrimSpace(reason)
 }
 
-// OpsStreamError 描述网关在「响应状态已固化为 200」之后（keepalive ping 或部分数据
-// 已 flush）就地以 SSE error 帧形式返回的错误。由于 HTTP 状态码停留在 200，
-// 而 ops_error_logger 以 status>=400 为采集触发条件，这类流内失败
-// （并发限流回退、Wait 后二次计费校验失败、流开始后才无可用账号等）本会在错误看板里
-// 完全隐形。handler.handleStreamingAwareError 负责标记，ops_error_logger 中间件在
-// status<400 分支消费它并补记一条错误日志。
+// OpsStreamError 描述承载在 2xx 响应上的带内错误：网关在响应状态已固化为 200 之后
+// 就地以 SSE error 帧返回的错误（并发限流回退、Wait 后二次计费校验失败、流开始后才无
+// 可用账号等），以及上游 2xx 正文或事件里携带的错误结果（NonStream 标记非流式正文）。
+// 由于 HTTP 状态码停留在 2xx，ops_error_logger 中间件在 status<400 分支消费该标记并
+// 补记错误日志；标记方是 handler.handleStreamingAwareError 或各 service 的带内检测。
 type OpsStreamError struct {
 	// ErrType 是写入 SSE 帧的对客错误类型（如 rate_limit_error / upstream_error / api_error）。
 	ErrType string
@@ -149,7 +148,7 @@ type OpsStreamError struct {
 	// Message 是写入 SSE 帧的对客错误消息。
 	Message string
 	// IntendedStatus 是流若未固化本应返回的 HTTP 状态码（如并发限流的 429）。
-	// 默认仅用于错误分级；CountTowardsSLA=true 时也作为 Ops 的逻辑状态码。
+	// 默认仅用于错误分级；CountTowardsSLA 或 RequestScoped 为 true 时也作为 Ops 的逻辑状态码。
 	IntendedStatus int
 	// CountTowardsSLA 表示虽然 wire 状态已固化为 200，请求在应用语义上仍然失败，
 	// Ops 应使用 IntendedStatus 计入错误率/SLA。
@@ -164,6 +163,12 @@ type OpsStreamError struct {
 	UpstreamMessage string
 	UpstreamDetail  string
 	UpstreamErrors  []*OpsUpstreamErrorEvent
+	// RequestScoped 表示该带内失败是请求级结果（如上游内容策略截停），与本请求此前的
+	// 上游尝试无关：分类不受上游错误上下文影响、不快照也不落库上游归因、不继承透传规则的
+	// skip_monitoring，按业务限制计，落库状态取 IntendedStatus 以便进入错误列表。
+	RequestScoped bool
+	// NonStream 表示带内信号来自非流式 2xx 响应体，落库 stream=false。
+	NonStream bool
 }
 
 const maxOpsStreamErrorsPerRequest = 64
@@ -206,6 +211,16 @@ func MarkOpsStreamFailure(c *gin.Context, errType, code, message string, intende
 	})
 }
 
+// MarkOpsStreamErrorValue 以完整的 OpsStreamError 记录一次带内错误，供需要
+// RequestScoped / NonStream 等附加语义的调用方使用；首个标记生效的规则不变。
+// 调用方只填 ErrType / Code / Message / IntendedStatus / CountTowardsSLA / RequestScoped / NonStream。
+// AccountID、UpstreamModel 与 Turn 由请求上下文接管；UpstreamStatus、UpstreamMessage、
+// UpstreamDetail、UpstreamErrors 与 SkipMonitoring 在非 RequestScoped 时由上下文接管，
+// RequestScoped 时被忽略。
+func MarkOpsStreamErrorValue(c *gin.Context, streamErr OpsStreamError) {
+	markOpsStreamError(c, streamErr)
+}
+
 func markOpsStreamError(c *gin.Context, streamErr OpsStreamError) {
 	if c == nil {
 		return
@@ -213,7 +228,9 @@ func markOpsStreamError(c *gin.Context, streamErr OpsStreamError) {
 	streamErr.ErrType = strings.TrimSpace(streamErr.ErrType)
 	streamErr.Code = strings.TrimSpace(streamErr.Code)
 	streamErr.Message = strings.TrimSpace(streamErr.Message)
-	streamErr.SkipMonitoring = currentOpsFailureSkipMonitoring(c)
+	if !streamErr.RequestScoped {
+		streamErr.SkipMonitoring = currentOpsFailureSkipMonitoring(c)
+	}
 	snapshotOpsStreamErrorContext(c, &streamErr)
 	if GetOpenAIClientTransport(c) == OpenAIClientTransportWS {
 		if value, ok := c.Get(OpsStreamTurnKey); ok {
@@ -253,6 +270,9 @@ func snapshotOpsStreamErrorContext(c *gin.Context, streamErr *OpsStreamError) {
 		streamErr.UpstreamModel, _ = value.(string)
 		streamErr.UpstreamModel = strings.TrimSpace(streamErr.UpstreamModel)
 	}
+	if streamErr.RequestScoped {
+		return
+	}
 	if value, ok := c.Get(OpsUpstreamStatusCodeKey); ok {
 		switch status := value.(type) {
 		case int: