Merge pull request #5118 from Wei-Shaw/fix/openai-proxy-stream-circuit-fail-open

fix(openai): 代理断流熔断改为 fail-open 偏好，修复共用代理部署下的调度不可用

backend/internal/config/config.go

@@ -1035,6 +1035,8 @@ type GatewayOpenAIHTTP2Config struct {
 // GatewayOpenAIProxyStreamCircuitConfig controls the bounded, in-process
 // proxy-ID circuit used for incomplete OpenAI Responses SSE streams.
 type GatewayOpenAIProxyStreamCircuitConfig struct {
+	// Disabled: 完全关闭代理断流熔断（默认开启）。
+	Disabled bool `mapstructure:"disabled"`
 	// FailureThreshold: 统计窗口内多少次断流后隔离代理。
 	FailureThreshold int `mapstructure:"failure_threshold"`
 	// WindowSeconds: 断流统计窗口（秒）。
@@ -2282,6 +2284,7 @@ func setDefaults() {
 	viper.SetDefault("gateway.openai_http2.fallback_error_threshold", 2)
 	viper.SetDefault("gateway.openai_http2.fallback_window_seconds", 60)
 	viper.SetDefault("gateway.openai_http2.fallback_ttl_seconds", 600)
+	viper.SetDefault("gateway.openai_proxy_stream_circuit.disabled", false)
 	viper.SetDefault("gateway.openai_proxy_stream_circuit.failure_threshold", 2)
 	viper.SetDefault("gateway.openai_proxy_stream_circuit.window_seconds", 60)
 	viper.SetDefault("gateway.openai_proxy_stream_circuit.ttl_seconds", 600)

backend/internal/config/config_test.go

@@ -545,6 +545,7 @@ func TestLoadDefaultOpenAIHTTP2Enabled(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, cfg.Gateway.OpenAIHTTP2.Enabled)
 	require.True(t, cfg.Gateway.OpenAIHTTP2.AllowProxyFallbackToHTTP1)
+	require.False(t, cfg.Gateway.OpenAIProxyStreamCircuit.Disabled)
 	require.Equal(t, 2, cfg.Gateway.OpenAIProxyStreamCircuit.FailureThreshold)
 	require.Equal(t, 60, cfg.Gateway.OpenAIProxyStreamCircuit.WindowSeconds)
 	require.Equal(t, 600, cfg.Gateway.OpenAIProxyStreamCircuit.TTLSeconds)
@@ -552,12 +553,14 @@ func TestLoadDefaultOpenAIHTTP2Enabled(t *testing.T) {
 
 func TestLoadOpenAIProxyStreamCircuitFromEnv(t *testing.T) {
 	resetViperWithJWTSecret(t)
+	t.Setenv("GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_DISABLED", "true")
 	t.Setenv("GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD", "3")
 	t.Setenv("GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS", "90")
 	t.Setenv("GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS", "420")
 
 	cfg, err := Load()
 	require.NoError(t, err)
+	require.True(t, cfg.Gateway.OpenAIProxyStreamCircuit.Disabled)
 	require.Equal(t, 3, cfg.Gateway.OpenAIProxyStreamCircuit.FailureThreshold)
 	require.Equal(t, 90, cfg.Gateway.OpenAIProxyStreamCircuit.WindowSeconds)
 	require.Equal(t, 420, cfg.Gateway.OpenAIProxyStreamCircuit.TTLSeconds)

backend/internal/service/openai_account_scheduler.go

@@ -3,6 +3,7 @@ package service
 import (
 	"container/heap"
 	"context"
+	"errors"
 	"fmt"
 	"hash/fnv"
 	"log/slog"
@@ -1678,7 +1679,7 @@ func (s *defaultOpenAIAccountScheduler) isAccountRequestCompatibleReason(ctx con
 	if s != nil && s.service != nil && s.service.isOpenAIAccountRequestRuntimeBlocked(account, req.RequestedModel) {
 		return false, "runtime_blocked"
 	}
-	if s != nil && s.service != nil && s.service.isOpenAIProxyStreamQuarantined(account) {
+	if s != nil && s.service != nil && s.service.isOpenAIProxyStreamQuarantined(ctx, account) {
 		return false, "proxy_stream_quarantined"
 	}
 	// Quota auto-pause must be evaluated during the initial filter too. Without it the
@@ -2035,6 +2036,13 @@ func (s *OpenAIGatewayService) SelectAccountWithSchedulerForImages(
 	return selection, decision, err
 }
 
+// selectAccountWithScheduler wraps selectAccountWithSchedulerOnce with a
+// fail-open second pass for the proxy stream circuit (#5056): when the only
+// reason no account is available is that every candidate sits behind a
+// quarantined proxy, the quarantine must degrade to a preference instead of
+// zeroing out capacity. The retry re-runs the exact same selection with the
+// quarantine checks bypassed, so healthy proxies always win the first pass
+// and quarantined ones only serve when nothing else can.
 func (s *OpenAIGatewayService) selectAccountWithScheduler(
 	ctx context.Context,
 	groupID *int64,
@@ -2049,6 +2057,40 @@ func (s *OpenAIGatewayService) selectAccountWithScheduler(
 	platform string,
 	previousResponseCanMove bool,
 	useUpstreamTokenCost bool,
+) (*AccountSelectionResult, OpenAIAccountScheduleDecision, error) {
+	selection, decision, err := s.selectAccountWithSchedulerOnce(ctx, groupID, previousResponseID, sessionHash, requestedModel, excludedIDs, requiredTransport, requiredCapability, requiredImageCapability, requireCompact, platform, previousResponseCanMove, useUpstreamTokenCost)
+	if err == nil || openAIProxyStreamQuarantineBypassed(ctx) {
+		return selection, decision, err
+	}
+	if !errors.Is(err, ErrNoAvailableAccounts) && !errors.Is(err, ErrNoAvailableCompactAccounts) {
+		return selection, decision, err
+	}
+	// The circuit only ever quarantines PlatformOpenAI accounts.
+	if normalizeOpenAICompatiblePlatform(platform) != PlatformOpenAI {
+		return selection, decision, err
+	}
+	blocked := s.getOpenAIProxyStreamCircuit().activeBlockCount(time.Now())
+	if blocked == 0 {
+		return selection, decision, err
+	}
+	s.logOpenAIProxyStreamQuarantineFailOpen(requestedModel, blocked)
+	return s.selectAccountWithSchedulerOnce(withOpenAIProxyStreamQuarantineBypass(ctx), groupID, previousResponseID, sessionHash, requestedModel, excludedIDs, requiredTransport, requiredCapability, requiredImageCapability, requireCompact, platform, previousResponseCanMove, useUpstreamTokenCost)
+}
+
+func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
+	ctx context.Context,
+	groupID *int64,
+	previousResponseID string,
+	sessionHash string,
+	requestedModel string,
+	excludedIDs map[int64]struct{},
+	requiredTransport OpenAIUpstreamTransport,
+	requiredCapability OpenAIEndpointCapability,
+	requiredImageCapability OpenAIImagesCapability,
+	requireCompact bool,
+	platform string,
+	previousResponseCanMove bool,
+	useUpstreamTokenCost bool,
 ) (*AccountSelectionResult, OpenAIAccountScheduleDecision, error) {
 	ctx = s.withOpenAIQuotaAutoPauseContext(ctx)
 	platform = normalizeOpenAICompatiblePlatform(platform)

backend/internal/service/openai_account_scheduler_test.go

@@ -1475,6 +1475,44 @@ func TestOpenAIGatewayService_SelectAccountWithScheduler_SkipsQuarantinedSharedP
 	require.Equal(t, int64(469803), selection.Account.ID)
 }
 
+func TestOpenAIGatewayService_SelectAccountWithScheduler_FailsOpenWhenAllProxiesQuarantined(t *testing.T) {
+	resetOpenAIAdvancedSchedulerSettingCacheForTest()
+
+	// Every schedulable account shares the quarantined proxy: the circuit must
+	// degrade to a preference instead of zeroing out capacity (#5056).
+	proxyA := int64(5056)
+	accounts := []Account{
+		{ID: 505601, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0, ProxyID: &proxyA},
+		{ID: 505602, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5, ProxyID: &proxyA},
+	}
+	cfg := &config.Config{}
+	cfg.Gateway.Scheduling.LoadBatchEnabled = false
+	svc := &OpenAIGatewayService{
+		accountRepo:        schedulerTestOpenAIAccountRepo{accounts: accounts},
+		cfg:                cfg,
+		concurrencyService: NewConcurrencyService(schedulerTestConcurrencyCache{}),
+		openaiProxyStreamCircuit: newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
+			failureThreshold: 1,
+			failureWindow:    time.Minute,
+			quarantineTTL:    10 * time.Minute,
+			maxEntries:       16,
+		}),
+	}
+	tripped, _ := svc.openaiProxyStreamCircuit.recordFailure(proxyA, time.Now())
+	require.True(t, tripped)
+
+	selection, _, err := svc.SelectAccountWithScheduler(
+		context.Background(), nil, "", "", "gpt-5.6-sol", nil, OpenAIUpstreamTransportAny, false,
+	)
+	require.NoError(t, err, "quarantine must fail open instead of returning no available accounts")
+	require.NotNil(t, selection)
+	require.NotNil(t, selection.Account)
+	require.NotNil(t, selection.Account.ProxyID)
+	require.Equal(t, proxyA, *selection.Account.ProxyID)
+	require.True(t, svc.openaiProxyStreamCircuit.isBlocked(proxyA, time.Now()),
+		"fail-open must not clear the quarantine; only a completed stream or TTL expiry does")
+}
+
 func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyRateLimitedAccountFallsBackToFreshCandidate(t *testing.T) {
 	ctx := context.Background()
 	groupID := int64(10101)

backend/internal/service/openai_gateway_scheduling.go

@@ -1243,7 +1243,7 @@ func (s *OpenAIGatewayService) resolveFreshSchedulableOpenAIAccount(ctx context.
 	if s.isOpenAIAccountRequestRuntimeBlocked(fresh, requestedModel) {
 		return nil
 	}
-	if s.isOpenAIProxyStreamQuarantined(fresh) {
+	if s.isOpenAIProxyStreamQuarantined(ctx, fresh) {
 		return nil
 	}
 	return fresh
@@ -1275,7 +1275,7 @@ func (s *OpenAIGatewayService) recheckSelectedOpenAIAccountFromDB(ctx context.Co
 		if !parentHealthyForShadow(account, s.parentAccountLookup(ctx)) {
 			return nil
 		}
-		if s.isOpenAIProxyStreamQuarantined(account) {
+		if s.isOpenAIProxyStreamQuarantined(ctx, account) {
 			return nil
 		}
 		return account
@@ -1297,7 +1297,7 @@ func (s *OpenAIGatewayService) recheckSelectedOpenAIAccountFromDB(ctx context.Co
 	if s.isOpenAIAccountRequestRuntimeBlocked(latest, requestedModel) {
 		return nil
 	}
-	if s.isOpenAIProxyStreamQuarantined(latest) {
+	if s.isOpenAIProxyStreamQuarantined(ctx, latest) {
 		return nil
 	}
 	return latest

backend/internal/service/openai_gateway_service.go

@@ -414,20 +414,21 @@ type OpenAIGatewayService struct {
 	liveAttestation       liveattestation.Provider
 	liveAttestationCipher SecretEncryptor
 
-	openaiWSPoolOnce              sync.Once
-	openaiWSStateStoreOnce        sync.Once
-	openaiSchedulerOnce           sync.Once
-	openaiProxyStreamCircuitOnce  sync.Once
-	openaiWSPassthroughDialerOnce sync.Once
-	openaiModelTransientOnce      sync.Once
-	agentIdentityTaskMu           sync.Mutex
-	openaiWSPool                  *openAIWSConnPool
-	openaiWSStateStore            OpenAIWSStateStore
-	openaiScheduler               OpenAIAccountScheduler
-	openaiWSPassthroughDialer     openAIWSClientDialer
-	openaiAccountStats            *openAIAccountRuntimeStats
-	openaiModelTransient          *openAIAccountModelTransientState
-	openaiProxyStreamCircuit      *openAIProxyStreamCircuit
+	openaiWSPoolOnce               sync.Once
+	openaiWSStateStoreOnce         sync.Once
+	openaiSchedulerOnce            sync.Once
+	openaiProxyStreamCircuitOnce   sync.Once
+	openaiWSPassthroughDialerOnce  sync.Once
+	openaiModelTransientOnce       sync.Once
+	agentIdentityTaskMu            sync.Mutex
+	openaiWSPool                   *openAIWSConnPool
+	openaiWSStateStore             OpenAIWSStateStore
+	openaiScheduler                OpenAIAccountScheduler
+	openaiWSPassthroughDialer      openAIWSClientDialer
+	openaiAccountStats             *openAIAccountRuntimeStats
+	openaiModelTransient           *openAIAccountModelTransientState
+	openaiProxyStreamCircuit       *openAIProxyStreamCircuit
+	openaiProxyStreamFailOpenLogAt atomic.Int64
 
 	openaiWSFallbackUntil               sync.Map // key: int64(accountID), value: time.Time
 	openaiAccountRuntimeBlockUntil      sync.Map // key: int64(accountID), value: time.Time

backend/internal/service/openai_gateway_service_test.go

@@ -1356,6 +1356,14 @@ func TestOpenAIStreamingPostOutputDisconnectQuarantinesSharedProxyWithoutSameStr
 	svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{
 		MaxLineSize: defaultMaxLineSize,
 	}}}
+	// collapseInterval 0: the two loop iterations below record within the
+	// production collapse window and must count as distinct failure events here.
+	svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
+		failureThreshold: 2,
+		failureWindow:    time.Minute,
+		quarantineTTL:    10 * time.Minute,
+		maxEntries:       16,
+	})
 
 	for _, readErr := range []error{
 		io.ErrUnexpectedEOF,
@@ -2008,6 +2016,14 @@ func TestOpenAIStreamingPassthroughPostOutputDisconnectQuarantinesSharedProxy(t
 	proxyID := int64(4698)
 	account := &Account{ID: 469804, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, ProxyID: &proxyID}
 	svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}}}
+	// collapseInterval 0: the loop below records within the production collapse
+	// window and must count as distinct failure events here.
+	svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
+		failureThreshold: 2,
+		failureWindow:    time.Minute,
+		quarantineTTL:    10 * time.Minute,
+		maxEntries:       16,
+	})
 
 	for _, readErr := range []error{io.ErrUnexpectedEOF, errors.New("http2: client connection lost")} {
 		rec := httptest.NewRecorder()
@@ -2029,7 +2045,7 @@ func TestOpenAIStreamingPassthroughPostOutputDisconnectQuarantinesSharedProxy(t
 		require.Contains(t, rec.Body.String(), "partial")
 	}
 
-	require.True(t, svc.isOpenAIProxyStreamQuarantined(account))
+	require.True(t, svc.isOpenAIProxyStreamQuarantined(context.Background(), account))
 }
 
 func TestOpenAIStreamingPassthroughResponseFailedBeforeOutputReturnsFailover(t *testing.T) {

backend/internal/service/openai_proxy_stream_circuit.go

@@ -15,20 +15,32 @@ const (
 	defaultOpenAIProxyStreamFailureWindow     = time.Minute
 	defaultOpenAIProxyStreamQuarantineTTL     = 10 * time.Minute
 	defaultOpenAIProxyStreamCircuitMaxEntries = 4096
+	// defaultOpenAIProxyStreamFailureCollapse merges disconnects that land
+	// within this interval into a single failure event. One proxy or HTTP/2
+	// connection loss kills every stream multiplexed on it at the same
+	// moment; counting each stream separately would trip the breaker from a
+	// single upstream event (#5056).
+	defaultOpenAIProxyStreamFailureCollapse = 3 * time.Second
+	// openAIProxyStreamFailOpenLogInterval rate-limits the fail-open warning
+	// so an outage-mode burst does not flood the log.
+	openAIProxyStreamFailOpenLogInterval = 5 * time.Second
 )
 
 type openAIProxyStreamCircuitSettings struct {
+	disabled         bool
 	failureThreshold int
 	failureWindow    time.Duration
 	quarantineTTL    time.Duration
+	collapseInterval time.Duration
 	maxEntries       int
 }
 
 type openAIProxyStreamCircuitEntry struct {
-	failureCount int
-	windowStart  time.Time
-	blockedUntil time.Time
-	lastTouched  time.Time
+	failureCount  int
+	windowStart   time.Time
+	lastFailureAt time.Time
+	blockedUntil  time.Time
+	lastTouched   time.Time
 }
 
 // openAIProxyStreamCircuit is an in-process, proxy-ID keyed circuit. It is
@@ -45,12 +57,14 @@ func resolveOpenAIProxyStreamCircuitSettings(s *OpenAIGatewayService) openAIProx
 		failureThreshold: defaultOpenAIProxyStreamFailureThreshold,
 		failureWindow:    defaultOpenAIProxyStreamFailureWindow,
 		quarantineTTL:    defaultOpenAIProxyStreamQuarantineTTL,
+		collapseInterval: defaultOpenAIProxyStreamFailureCollapse,
 		maxEntries:       defaultOpenAIProxyStreamCircuitMaxEntries,
 	}
 	if s == nil || s.cfg == nil {
 		return settings
 	}
 	cfg := s.cfg.Gateway.OpenAIProxyStreamCircuit
+	settings.disabled = cfg.Disabled
 	if cfg.FailureThreshold > 0 {
 		settings.failureThreshold = cfg.FailureThreshold
 	}
@@ -76,6 +90,9 @@ func newOpenAIProxyStreamCircuit(settings openAIProxyStreamCircuitSettings) *ope
 	if settings.maxEntries <= 0 {
 		settings.maxEntries = defaultOpenAIProxyStreamCircuitMaxEntries
 	}
+	if settings.collapseInterval < 0 {
+		settings.collapseInterval = 0
+	}
 	return &openAIProxyStreamCircuit{
 		settings: settings,
 		entries:  make(map[int64]openAIProxyStreamCircuitEntry),
@@ -95,7 +112,7 @@ func (s *OpenAIGatewayService) getOpenAIProxyStreamCircuit() *openAIProxyStreamC
 }
 
 func (c *openAIProxyStreamCircuit) recordFailure(proxyID int64, now time.Time) (bool, time.Time) {
-	if c == nil || proxyID <= 0 {
+	if c == nil || c.settings.disabled || proxyID <= 0 {
 		return false, time.Time{}
 	}
 	c.mu.Lock()
@@ -115,7 +132,17 @@ func (c *openAIProxyStreamCircuit) recordFailure(proxyID int64, now time.Time) (
 		entry.windowStart = now
 		entry.blockedUntil = time.Time{}
 	}
+	// Collapse a burst of disconnects into one failure event: when a proxy or
+	// a multiplexed HTTP/2 connection dies, every in-flight stream reports the
+	// same underlying incident within moments of each other.
+	if c.settings.collapseInterval > 0 && !entry.lastFailureAt.IsZero() &&
+		now.Sub(entry.lastFailureAt) >= 0 && now.Sub(entry.lastFailureAt) < c.settings.collapseInterval {
+		entry.lastTouched = now
+		c.entries[proxyID] = entry
+		return false, time.Time{}
+	}
 	entry.failureCount++
+	entry.lastFailureAt = now
 	entry.lastTouched = now
 	tripped := entry.failureCount >= c.settings.failureThreshold
 	if tripped {
@@ -139,7 +166,7 @@ func (c *openAIProxyStreamCircuit) recordSuccess(proxyID int64) bool {
 }
 
 func (c *openAIProxyStreamCircuit) isBlocked(proxyID int64, now time.Time) bool {
-	if c == nil || proxyID <= 0 {
+	if c == nil || c.settings.disabled || proxyID <= 0 {
 		return false
 	}
 	c.mu.Lock()
@@ -155,6 +182,25 @@ func (c *openAIProxyStreamCircuit) isBlocked(proxyID int64, now time.Time) bool
 	return true
 }
 
+// activeBlockCount reports how many proxies are currently quarantined. It
+// gates the fail-open retry: a "no available accounts" selection result only
+// warrants a second, quarantine-blind pass when the circuit is actually
+// withholding capacity.
+func (c *openAIProxyStreamCircuit) activeBlockCount(now time.Time) int {
+	if c == nil || c.settings.disabled {
+		return 0
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	count := 0
+	for _, entry := range c.entries {
+		if !entry.blockedUntil.IsZero() && now.Before(entry.blockedUntil) {
+			count++
+		}
+	}
+	return count
+}
+
 func (c *openAIProxyStreamCircuit) ensureCapacityLocked(now time.Time) {
 	if len(c.entries) < c.settings.maxEntries {
 		return
@@ -219,11 +265,48 @@ func (s *OpenAIGatewayService) clearOpenAIProxyStreamDisconnect(account *Account
 	}
 }
 
-func (s *OpenAIGatewayService) isOpenAIProxyStreamQuarantined(account *Account) bool {
+// openAIProxyStreamQuarantineBypassKey marks a selection pass that must ignore
+// proxy quarantine. It is set for the second, fail-open selection attempt when
+// the first pass found no available account while the circuit was withholding
+// proxies: a degraded proxy is strictly better than answering 502 (#5056).
+type openAIProxyStreamQuarantineBypassKey struct{}
+
+func withOpenAIProxyStreamQuarantineBypass(ctx context.Context) context.Context {
+	return context.WithValue(ctx, openAIProxyStreamQuarantineBypassKey{}, true)
+}
+
+func openAIProxyStreamQuarantineBypassed(ctx context.Context) bool {
+	if ctx == nil {
+		return false
+	}
+	bypassed, _ := ctx.Value(openAIProxyStreamQuarantineBypassKey{}).(bool)
+	return bypassed
+}
+
+func (s *OpenAIGatewayService) isOpenAIProxyStreamQuarantined(ctx context.Context, account *Account) bool {
 	proxyID, ok := openAIProxyStreamCircuitProxyID(account)
 	if !ok {
 		return false
 	}
+	if openAIProxyStreamQuarantineBypassed(ctx) {
+		return false
+	}
 	circuit := s.getOpenAIProxyStreamCircuit()
 	return circuit != nil && circuit.isBlocked(proxyID, time.Now())
 }
+
+// logOpenAIProxyStreamQuarantineFailOpen emits a rate-limited warning when a
+// selection pass had to re-admit quarantined proxies to serve at all.
+func (s *OpenAIGatewayService) logOpenAIProxyStreamQuarantineFailOpen(requestedModel string, blockedProxies int) {
+	now := time.Now().UnixNano()
+	last := s.openaiProxyStreamFailOpenLogAt.Load()
+	if now-last < int64(openAIProxyStreamFailOpenLogInterval) ||
+		!s.openaiProxyStreamFailOpenLogAt.CompareAndSwap(last, now) {
+		return
+	}
+	logger.L().With(zap.String("component", "service.openai_gateway")).Warn(
+		"openai.proxy_stream_quarantine_fail_open",
+		zap.Int("blocked_proxies", blockedProxies),
+		zap.String("model", requestedModel),
+	)
+}

backend/internal/service/openai_proxy_stream_circuit_test.go

@@ -1,6 +1,7 @@
 package service
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -35,6 +36,82 @@ func TestOpenAIProxyStreamCircuitThresholdTTLAndSuccessReset(t *testing.T) {
 	require.False(t, tripped, "failures outside the window must not accumulate")
 }
 
+func TestOpenAIProxyStreamCircuitCollapsesBurstFailures(t *testing.T) {
+	base := time.Unix(1_800_000_000, 0)
+	circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
+		failureThreshold: 2,
+		failureWindow:    time.Minute,
+		quarantineTTL:    10 * time.Minute,
+		collapseInterval: 3 * time.Second,
+		maxEntries:       16,
+	})
+
+	// One HTTP/2 connection loss kills several multiplexed streams at once:
+	// the near-simultaneous reports must count as a single failure event.
+	tripped, _ := circuit.recordFailure(1, base)
+	require.False(t, tripped)
+	tripped, _ = circuit.recordFailure(1, base.Add(time.Second))
+	require.False(t, tripped, "burst failures inside the collapse interval must merge")
+	tripped, _ = circuit.recordFailure(1, base.Add(2*time.Second))
+	require.False(t, tripped, "burst failures inside the collapse interval must merge")
+	require.False(t, circuit.isBlocked(1, base.Add(2*time.Second)))
+
+	// A second, distinct incident past the collapse interval still trips.
+	tripped, _ = circuit.recordFailure(1, base.Add(5*time.Second))
+	require.True(t, tripped)
+	require.True(t, circuit.isBlocked(1, base.Add(5*time.Second)))
+}
+
+func TestOpenAIProxyStreamCircuitDisabled(t *testing.T) {
+	base := time.Unix(1_800_000_000, 0)
+	circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
+		disabled:         true,
+		failureThreshold: 1,
+		failureWindow:    time.Minute,
+		quarantineTTL:    10 * time.Minute,
+		maxEntries:       16,
+	})
+
+	tripped, _ := circuit.recordFailure(1, base)
+	require.False(t, tripped)
+	require.False(t, circuit.isBlocked(1, base))
+	require.Equal(t, 0, circuit.activeBlockCount(base))
+}
+
+func TestOpenAIProxyStreamCircuitActiveBlockCount(t *testing.T) {
+	base := time.Unix(1_800_000_000, 0)
+	circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
+		failureThreshold: 1,
+		failureWindow:    time.Minute,
+		quarantineTTL:    10 * time.Minute,
+		maxEntries:       16,
+	})
+
+	require.Equal(t, 0, circuit.activeBlockCount(base))
+	tripped, until := circuit.recordFailure(1, base)
+	require.True(t, tripped)
+	circuit.recordFailure(2, base) // second proxy also tripped (threshold 1)
+	require.Equal(t, 2, circuit.activeBlockCount(base.Add(time.Second)))
+	require.Equal(t, 0, circuit.activeBlockCount(until), "expired quarantines must not count")
+}
+
+func TestOpenAIProxyStreamQuarantineBypassContext(t *testing.T) {
+	proxyID := int64(7)
+	account := &Account{ID: 1, Platform: PlatformOpenAI, ProxyID: &proxyID}
+	svc := &OpenAIGatewayService{}
+	svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{
+		failureThreshold: 1,
+		failureWindow:    time.Minute,
+		quarantineTTL:    10 * time.Minute,
+		maxEntries:       16,
+	})
+	svc.openaiProxyStreamCircuit.recordFailure(proxyID, time.Now())
+
+	ctx := context.Background()
+	require.True(t, svc.isOpenAIProxyStreamQuarantined(ctx, account))
+	require.False(t, svc.isOpenAIProxyStreamQuarantined(withOpenAIProxyStreamQuarantineBypass(ctx), account))
+}
+
 func TestOpenAIProxyStreamCircuitBoundsEntries(t *testing.T) {
 	base := time.Unix(1_800_000_000, 0)
 	circuit := newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{

deploy/config.example.yaml

@@ -423,7 +423,10 @@ gateway:
     fallback_window_seconds: 60
     fallback_ttl_seconds: 600
   # OpenAI Responses SSE 代理断流熔断；按 proxy_id 跨账号隔离，仅影响下一次调度。
+  # 隔离是偏好而非硬闸：当全部候选账号都在被隔离代理下时自动放行（fail-open），
+  # 不会因熔断而返回"无可用账号"。3 秒内的并发断流合并为一次失败事件。
   openai_proxy_stream_circuit:
+    disabled: false
     failure_threshold: 2
     window_seconds: 60
     ttl_seconds: 600