fix(openai): track HTTP/2 stream body failures

backend/internal/repository/http_upstream.go

@@ -136,6 +136,7 @@ type upstreamClientEntry struct {
 type openAIHTTP2FallbackState struct {
 	mu            sync.Mutex
 	windowStart   time.Time
+	lastFailureAt time.Time
 	errorCount    int
 	fallbackUntil time.Time
 }
@@ -213,6 +214,7 @@ func (s *httpUpstreamService) Do(req *http.Request, proxyURL string, accountID i
 	}
 
 	// 执行请求
+	requestStartedAt := time.Now()
 	client := httpClientForUpstreamRequest(entry.client, req)
 	client = httpClientWithGrokAccessDeniedFallback(client)
 	resp, err := servertiming.Do(client, req)
@@ -223,17 +225,23 @@ func (s *httpUpstreamService) Do(req *http.Request, proxyURL string, accountID i
 		atomic.StoreInt64(&entry.lastUsed, time.Now().UnixNano())
 		return nil, err
 	}
-	s.recordOpenAIHTTP2Success(profile, entry.protocolMode, entry.proxyKey)
 
 	// 如果上游返回了压缩内容，解压后再交给业务层
 	decompressResponseBody(resp)
 
-	// 包装响应体，在关闭时自动减少计数并更新时间戳
+	// H2 streams can fail after a successful response header. Classify the whole
+	// body lifecycle so repeated mid-stream failures can activate the existing
+	// per-proxy H2 -> H1 compatibility fallback instead of being reset by 200 headers.
 	// 这确保了流式响应（如 SSE）在完全读取前不会被淘汰
-	resp.Body = wrapTrackedBody(resp.Body, func() {
-		atomic.AddInt64(&entry.inFlight, -1)
-		atomic.StoreInt64(&entry.lastUsed, time.Now().UnixNano())
-	})
+	resp.Body = wrapTrackedBodyWithOutcome(
+		resp.Body,
+		func() {
+			atomic.AddInt64(&entry.inFlight, -1)
+			atomic.StoreInt64(&entry.lastUsed, time.Now().UnixNano())
+		},
+		func() { s.recordOpenAIHTTP2Success(profile, entry.protocolMode, entry.proxyKey, requestStartedAt) },
+		func(err error) { s.recordOpenAIHTTP2Failure(profile, entry.protocolMode, entry.proxyKey, err) },
+	)
 
 	return resp, nil
 }
@@ -1057,6 +1065,7 @@ func isOpenAIHTTP2CompatibilityError(err error) bool {
 		"no application protocol",
 		"protocol error",
 		"stream error",
+		"http2: client connection lost",
 		"goaway",
 		"refused_stream",
 		"frame too large",
@@ -1119,7 +1128,12 @@ func (s *httpUpstreamService) recordOpenAIHTTP2Failure(profile service.HTTPUpstr
 	}
 }
 
-func (s *httpUpstreamService) recordOpenAIHTTP2Success(profile service.HTTPUpstreamProfile, protocolMode, proxyKey string) {
+func (s *httpUpstreamService) recordOpenAIHTTP2Success(
+	profile service.HTTPUpstreamProfile,
+	protocolMode string,
+	proxyKey string,
+	requestStartedAt time.Time,
+) {
 	if profile != service.HTTPUpstreamProfileOpenAI || protocolMode != upstreamProtocolModeOpenAIH2 {
 		return
 	}
@@ -1134,7 +1148,7 @@ func (s *httpUpstreamService) recordOpenAIHTTP2Success(profile service.HTTPUpstr
 	if !ok || state == nil {
 		return
 	}
-	state.resetErrorWindow()
+	state.resetErrorWindow(requestStartedAt)
 }
 
 func (s *openAIHTTP2FallbackState) isFallbackActive(now time.Time) bool {
@@ -1150,10 +1164,14 @@ func (s *openAIHTTP2FallbackState) isFallbackActive(now time.Time) bool {
 	return false
 }
 
-func (s *openAIHTTP2FallbackState) resetErrorWindow() {
+func (s *openAIHTTP2FallbackState) resetErrorWindow(requestStartedAt time.Time) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if !s.lastFailureAt.IsZero() && requestStartedAt.Before(s.lastFailureAt) {
+		return
+	}
 	s.windowStart = time.Time{}
+	s.lastFailureAt = time.Time{}
 	s.errorCount = 0
 }
 
@@ -1183,6 +1201,7 @@ func (s *openAIHTTP2FallbackState) recordFailure(now time.Time, threshold int, w
 		s.errorCount = 0
 	}
 	s.errorCount++
+	s.lastFailureAt = now
 	if s.errorCount < threshold {
 		return false, time.Time{}
 	}
@@ -1422,18 +1441,49 @@ type trackedBody struct {
 	io.ReadCloser // 原始响应体
 	once          sync.Once
 	onClose       func() // 关闭时的回调函数
+	outcomeOnce   sync.Once
+	onSuccess     func()
+	onReadError   func(error)
+}
+
+func (b *trackedBody) Read(p []byte) (int, error) {
+	n, err := b.ReadCloser.Read(p)
+	if err == nil {
+		return n, nil
+	}
+	if errors.Is(err, io.EOF) {
+		b.recordSuccess()
+	} else {
+		b.recordReadError(err)
+	}
+	return n, err
 }
 
 // Close 关闭响应体并执行回调
 // 使用 sync.Once 确保回调只执行一次
 func (b *trackedBody) Close() error {
 	err := b.ReadCloser.Close()
+	if err != nil {
+		b.recordReadError(err)
+	}
 	if b.onClose != nil {
 		b.once.Do(b.onClose)
 	}
 	return err
 }
 
+func (b *trackedBody) recordSuccess() {
+	if b.onSuccess != nil {
+		b.outcomeOnce.Do(b.onSuccess)
+	}
+}
+
+func (b *trackedBody) recordReadError(err error) {
+	if b.onReadError != nil {
+		b.outcomeOnce.Do(func() { b.onReadError(err) })
+	}
+}
+
 // wrapTrackedBody 包装响应体以跟踪关闭事件
 // 用于在响应体关闭时更新 inFlight 计数
 //
@@ -1444,10 +1494,24 @@ func (b *trackedBody) Close() error {
 // 返回:
 //   - io.ReadCloser: 包装后的响应体
 func wrapTrackedBody(body io.ReadCloser, onClose func()) io.ReadCloser {
+	return wrapTrackedBodyWithOutcome(body, onClose, nil, nil)
+}
+
+func wrapTrackedBodyWithOutcome(
+	body io.ReadCloser,
+	onClose func(),
+	onSuccess func(),
+	onReadError func(error),
+) io.ReadCloser {
 	if body == nil {
 		return body
 	}
-	return &trackedBody{ReadCloser: body, onClose: onClose}
+	return &trackedBody{
+		ReadCloser:  body,
+		onClose:     onClose,
+		onSuccess:   onSuccess,
+		onReadError: onReadError,
+	}
 }
 
 // decompressResponseBody 根据 Content-Encoding 解压响应体。

backend/internal/repository/http_upstream_test.go

@@ -731,6 +731,134 @@ func (s *HTTPUpstreamSuite) TestOpenAIHTTP2ProxyCompatibilityErrorActivatesFallb
 	require.Equal(s.T(), upstreamProtocolModeOpenAIH1Fallback, entry.protocolMode)
 }
 
+func (s *HTTPUpstreamSuite) TestOpenAIHTTP2BodyFailuresActivateProxyFallback() {
+	s.cfg.Gateway = config.GatewayConfig{
+		OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
+			Enabled:                   true,
+			AllowProxyFallbackToHTTP1: true,
+			FallbackErrorThreshold:    2,
+			FallbackWindowSeconds:     60,
+			FallbackTTLSeconds:        600,
+		},
+	}
+	svc := s.newService()
+	proxyURL := "http://proxy.local:8080"
+	readBody := func(readErr error) {
+		startedAt := time.Now()
+		body := wrapTrackedBodyWithOutcome(
+			io.NopCloser(&failingReader{err: readErr}),
+			nil,
+			func() {
+				svc.recordOpenAIHTTP2Success(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, startedAt)
+			},
+			func(err error) {
+				svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, err)
+			},
+		)
+		_, err := io.ReadAll(body)
+		require.Error(s.T(), err)
+		require.NoError(s.T(), body.Close())
+	}
+
+	readBody(errors.New("http2: client connection lost"))
+	require.False(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL))
+	readBody(errors.New("http2: client connection lost"))
+	require.True(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL))
+}
+
+func (s *HTTPUpstreamSuite) TestOpenAIHTTP2GenericUnexpectedEOFDoesNotActivateProxyFallback() {
+	s.cfg.Gateway = config.GatewayConfig{
+		OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
+			Enabled:                   true,
+			AllowProxyFallbackToHTTP1: true,
+			FallbackErrorThreshold:    1,
+			FallbackWindowSeconds:     60,
+			FallbackTTLSeconds:        600,
+		},
+	}
+	svc := s.newService()
+	proxyURL := "http://proxy.local:8080"
+	body := wrapTrackedBodyWithOutcome(
+		io.NopCloser(&failingReader{err: io.ErrUnexpectedEOF}),
+		nil,
+		nil,
+		func(err error) {
+			svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, err)
+		},
+	)
+
+	_, err := io.ReadAll(body)
+	require.ErrorIs(s.T(), err, io.ErrUnexpectedEOF)
+	require.NoError(s.T(), body.Close())
+	require.False(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL))
+}
+
+func (s *HTTPUpstreamSuite) TestOpenAIHTTP2SuccessfulBodyResetsFailureWindow() {
+	s.cfg.Gateway = config.GatewayConfig{
+		OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
+			Enabled:                   true,
+			AllowProxyFallbackToHTTP1: true,
+			FallbackErrorThreshold:    2,
+			FallbackWindowSeconds:     60,
+			FallbackTTLSeconds:        600,
+		},
+	}
+	svc := s.newService()
+	proxyURL := "http://proxy.local:8080"
+	failed := func() {
+		startedAt := time.Now()
+		body := wrapTrackedBodyWithOutcome(
+			io.NopCloser(&failingReader{err: errors.New("http2: client connection lost")}),
+			nil,
+			func() {
+				svc.recordOpenAIHTTP2Success(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, startedAt)
+			},
+			func(err error) {
+				svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, err)
+			},
+		)
+		_, err := io.ReadAll(body)
+		require.Error(s.T(), err)
+		require.NoError(s.T(), body.Close())
+	}
+	succeeded := func() {
+		startedAt := time.Now()
+		body := wrapTrackedBodyWithOutcome(
+			io.NopCloser(strings.NewReader("ok")),
+			nil,
+			func() {
+				svc.recordOpenAIHTTP2Success(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, startedAt)
+			},
+			func(err error) {
+				svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, err)
+			},
+		)
+		_, err := io.ReadAll(body)
+		require.NoError(s.T(), err)
+		require.NoError(s.T(), body.Close())
+	}
+
+	failed()
+	succeeded()
+	failed()
+	require.False(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL))
+}
+
+func (s *HTTPUpstreamSuite) TestOpenAIHTTP2OlderSuccessCannotResetNewFailure() {
+	base := time.Unix(1_800_000_000, 0)
+	state := &openAIHTTP2FallbackState{}
+
+	tripped, _ := state.recordFailure(base, 2, time.Minute, 10*time.Minute)
+	require.False(s.T(), tripped)
+	state.resetErrorWindow(base.Add(-time.Second))
+	tripped, _ = state.recordFailure(base.Add(10*time.Second), 2, time.Minute, 10*time.Minute)
+	require.True(s.T(), tripped, "a success from an older request must not erase a newer transport failure")
+}
+
+type failingReader struct{ err error }
+
+func (r *failingReader) Read([]byte) (int, error) { return 0, r.err }
+
 // TestNormalizeProxyURL_Canonicalizes 测试代理 URL 规范化
 // 验证等价地址能够映射到同一缓存键
 func (s *HTTPUpstreamSuite) TestNormalizeProxyURL_Canonicalizes() {

backend/internal/service/openai_proxy_stream_circuit.go

@@ -158,7 +158,11 @@ func (c *openAIProxyStreamCircuit) recordSuccess(proxyID int64) bool {
 	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	if _, ok := c.entries[proxyID]; !ok {
+	entry, ok := c.entries[proxyID]
+	if !ok || !entry.blockedUntil.IsZero() {
+		// A success can belong to a request that started before a concurrent
+		// failure tripped the circuit. Once quarantined, keep the proxy blocked
+		// for the configured TTL; fail-open selection still preserves capacity.
 		return false
 	}
 	delete(c.entries, proxyID)

backend/internal/service/openai_proxy_stream_circuit_test.go

@@ -28,6 +28,8 @@ func TestOpenAIProxyStreamCircuitThresholdTTLAndSuccessReset(t *testing.T) {
 	require.True(t, tripped)
 	require.Equal(t, base.Add(20*time.Second+10*time.Minute), until)
 	require.True(t, circuit.isBlocked(1, until.Add(-time.Nanosecond)))
+	require.False(t, circuit.recordSuccess(1), "an older in-flight success must not clear an active quarantine")
+	require.True(t, circuit.isBlocked(1, until.Add(-time.Nanosecond)))
 	require.False(t, circuit.isBlocked(1, until), "TTL expiry must re-admit the proxy")
 
 	tripped, _ = circuit.recordFailure(2, base)