feat: suppress consecutive-channel opening on evidently healthy sources

Round 8 forensics caught the healthy source circuit-opened by five
random empty completions (~20% ambient failure rate makes a 5-streak
land every ~3000 attempts), blacking out the only good source for 60s.
When the window holds min_calls samples below the failure-rate
threshold, a streak is noise and no longer opens the gate; cold-start
and low-traffic semantics are unchanged and sudden death of a warm
source is still caught by the rate channel.

research-wiki/designs/2026-07-21-m25-resilience-design.md

@@ -110,6 +110,11 @@
 **动因**: 第六轮实测网关限速是**按账号速率**而非并发——外部负载高峰时我方仅 26 req/min 仍吃 16% 429,AIMD(并发型)压不住速率型限流;429 每次消耗 1/3 重试预算,饱和窗口里调用被"429 链"干净杀死(16 retry_exhausted/336)。
 **修正**: 携 Retry-After 的 429 是**服务端调度指令而非失败**(gRPC A6 pushback / SRE 语义): 照常退避(与 Retry-After 取大)、照常喂 AIMD/健康分,但**不消耗重试预算**;新增**调用级时间上限**(复用 `stall_window_s`,缺省 300s)在重试循环顶部兜底——持续饱和超窗即按既有 `stalled` 语义抛出,防无限循环。对下游的承诺变化: 饱和期调用最长等待 stall_window 而非快速失败(生产语义,迁移文档标注)。
 
+### 3.39 迭代 6 补遗: 健康证据抑制连续通道(2026-07-21,第八轮取证驱动)
+
+**动因**: 第八轮失败链取证发现健康源1 自己进 cooldown——空补全率 ~20% 下 5 连败每 ~3000 次尝试随机发生一次,连续通道(阈值 5)把唯一好源关 60s,期间调用全灭(链样本: `{'minimax_1': 'cooldown', 其余: 'cooldown/rate_limited'}`)。
+**修正**(Resilience4j 精神: 率证据优于连败直觉): 窗口样本 ≥ min_calls 且失败率 < fail_rate 时**抑制连续通道开路**——有充分健康证据的源上连败是统计噪声。连续通道本职(冷启动/低流量快杀死源)保留(窗口样本不足时照常开路);温热源猝死由率通道在失败累积后接管(检出延迟从 threshold 次升至率窗口收敛,契约测试⑩钉住);SourceDead force_open 不受影响。
+
 ### 3.4 不做与预留(方案 C 组件的接入点)
 
 - 账号级 429 共享退避: 不做;预留 = 冷却备忘 key 从 source_name 换 account_key 即可接入(LiteLLM 先例,治理粒度=配额粒度原则记入 ARCHITECTURE)。

src/polygateway/backends/memory/breaker.py

@@ -81,6 +81,14 @@ class InMemoryGate:
         if failed:
             g.f0 += 1
 
+    def _window_evidently_healthy(self, g: _SourceGate) -> bool:
+        """窗口样本充足且失败率低于阈值 = 有充分健康证据(迭代 6 抑制判据)。"""
+        self._rotate_window(g)
+        attempts = g.a0 + g.a1
+        if attempts < self._cfg.min_calls:
+            return False
+        return (g.f0 + g.f1) / attempts < self._cfg.fail_rate
+
     def _rate_channel_open(self, g: _SourceGate) -> bool:
         self._rotate_window(g)
         attempts = g.a0 + g.a1
@@ -227,8 +235,10 @@ class InMemoryGate:
         g.fails += 1
         if self._rate_channel_open(g):
             self._open(g, reason, bump_streak=True)
-        elif g.fails >= self._cfg.fail_threshold:
+        elif g.fails >= self._cfg.fail_threshold and not self._window_evidently_healthy(g):
-            self._open(g, reason, bump_streak=False)  # 连续通道不递增(C1 封顶)
+            # 连续通道不递增(C1 封顶);窗口证据充足且健康时连败是噪声,
+            # 不误熔"当前最好的源"(迭代 6,第八轮实证: 源1 被 5 连空补全误关 60s)
+            self._open(g, reason, bump_streak=False)
         return self._snapshot(g, applied=True)
 
     async def release_probe(self, entry: GateDecision) -> GateUpdate:

src/polygateway/backends/redis/breaker.py

@@ -179,6 +179,7 @@ end
 
 local threshold = tonumber(ARGV[5])
 local open_via_rate = false
+local window_healthy = false
 if force_open == 0 and is_probe == 0 then
     rotate_window(KEYS[1], now, tonumber(ARGV[10]))
     redis.call('HINCRBY', KEYS[1], 'a0', 1)
@@ -187,8 +188,13 @@ if force_open == 0 and is_probe == 0 then
         + tonumber(redis.call('HGET', KEYS[1], 'a1') or '0')
     local fails_w = tonumber(redis.call('HGET', KEYS[1], 'f0') or '0')
         + tonumber(redis.call('HGET', KEYS[1], 'f1') or '0')
-    if attempts >= tonumber(ARGV[8]) and fails_w / attempts >= tonumber(ARGV[9]) then
+    if attempts >= tonumber(ARGV[8]) then
+        if fails_w / attempts >= tonumber(ARGV[9]) then
             open_via_rate = true
+        else
+            -- 窗口证据充足且健康: 连败是噪声,抑制连续通道(迭代 6)
+            window_healthy = true
+        end
     end
 end
 
@@ -197,7 +203,8 @@ if force_open == 1 or is_probe == 1 then
 else
     failures = failures + 1
 end
-if force_open == 1 or is_probe == 1 or open_via_rate or failures >= threshold then
+local open_via_streak = failures >= threshold and not window_healthy
+if force_open == 1 or is_probe == 1 or open_via_rate or open_via_streak then
     -- 递增 streak 的只有率通道开路与探针失败重开(C1: 连续通道/force_open 不递增)
     if open_via_rate or is_probe == 1 then
         redis.call('HINCRBY', KEYS[1], 'reopen_streak', 1)

tests/contracts/test_breaker_contract.py

@@ -290,3 +290,36 @@ class TestRetryAfter:
         gate = gate_factory(_CFG)
         await _open_gate(gate, "s1")  # s1 开路;s2 健康
         assert await gate.retry_after_s(("s1", "s2")) == 0.0
+
+
+class TestConsecutiveSuppression:
+    """迭代 6: 窗口证据充足且健康时,连败是噪声,不开路(设计 §3.39)。"""
+
+    async def test_streak_suppressed_on_evidently_healthy_source(self, gate_factory):
+        # 20 成功垫底(窗口样本充足、失败率低)后 3 连败(阈值 3)→ 不开路
+        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=10)
+        gate = gate_factory(cfg)
+        for _ in range(20):
+            entry = await gate.try_enter("s1", "w")
+            await gate.record_success(entry)
+        await _fail(gate, reason="timeout", n=3)
+        assert (await gate.try_enter("s1", "w")).allowed
+
+    async def test_streak_fires_when_window_insufficient(self, gate_factory):
+        # 冷启动(窗口样本不足)3 连败照常开路——连续通道本职保留
+        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=10)
+        gate = gate_factory(cfg)
+        update = await _fail(gate, reason="timeout", n=3)
+        assert update.state is GateState.OPEN
+
+    async def test_warm_source_going_dead_caught_by_rate(self, gate_factory):
+        # 温热源猝死: 连败被抑制,但失败率窗口随失败累积必然接管
+        cfg = BreakerConfig(
+            fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=4, fail_rate=0.6
+        )
+        gate = gate_factory(cfg)
+        for _ in range(3):
+            entry = await gate.try_enter("s1", "w")
+            await gate.record_success(entry)
+        update = await _fail(gate, reason="timeout", n=5)  # 5/8 = 0.625 ≥ 0.6
+        assert update.state is GateState.OPEN