diff --git a/src/polygateway/backends/memory/breaker.py b/src/polygateway/backends/memory/breaker.py
index e9f3fbf..72df5c4 100644
--- a/src/polygateway/backends/memory/breaker.py
+++ b/src/polygateway/backends/memory/breaker.py
@@ -4,6 +4,12 @@
 契约形态承 CHS `provider_gate.py`: 半开探针是**带 TTL 的租约**(持有者
 死亡后可被接管,防"探针永远在路上"死锁),写回经 epoch fencing 拒绝
 旧世代污染。epoch 在每次进入 OPEN 时递增。时钟构造注入,纯确定性可测。
+
+M2.5 双通道(设计 2026-07-21-m25 §3.1): 在连续失败通道之外加失败率
+通道(双 30s 桶窗口,样本 ≥ min_calls 且失败率 ≥ fail_rate 即开路);
+429(rate_limited)是背压不是故障,两通道均不计;开路时长按 reopen_streak
+指数递增封顶 max_cooldown_s——连续通道触发的开路不递增 streak(误熔
+健康源的代价封顶为单次 cooldown_s)。
 """
 
 from __future__ import annotations
@@ -22,7 +28,7 @@ if TYPE_CHECKING:
 
 @dataclass
 class _SourceGate:
-    """单源的门控可变状态。"""
+    """单源的门控可变状态(窗口/退避域见设计 §3.1 状态机域清单)。"""
 
     state: GateState = GateState.CLOSED
     fails: int = 0
@@ -31,6 +37,14 @@ class _SourceGate:
     probe_owner: str | None = None
     probe_expires: float = 0.0
     reasons: dict[str, str] = field(default_factory=dict)
+    # 失败率窗口: 双半窗桶轮换(win_id = now // (window_s/2))
+    win_id: int = -1
+    a0: int = 0  # 当前桶 attempts / failures
+    f0: int = 0
+    a1: int = 0  # 上一桶
+    f1: int = 0
+    reopen_streak: int = 0
+    closed_since: float | None = None
 
 
 class InMemoryGate:
@@ -46,6 +60,38 @@ class InMemoryGate:
             raise ValueError("source_name 不能为空")
         return self._gates.setdefault(source_name, _SourceGate())
 
+    # —— M2.5 失败率窗口 ——
+
+    def _rotate_window(self, g: _SourceGate) -> None:
+        """按半窗粒度轮换双桶;跨两桶以上的空窗直接清零。"""
+        half = self._cfg.window_s / 2.0
+        wid = int(self._now() // half)
+        if wid == g.win_id:
+            return
+        if wid == g.win_id + 1:
+            g.a1, g.f1 = g.a0, g.f0
+        else:
+            g.a1, g.f1 = 0, 0
+        g.a0, g.f0 = 0, 0
+        g.win_id = wid
+
+    def _window_add(self, g: _SourceGate, *, failed: bool) -> None:
+        self._rotate_window(g)
+        g.a0 += 1
+        if failed:
+            g.f0 += 1
+
+    def _rate_channel_open(self, g: _SourceGate) -> bool:
+        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 _cooldown_eff(self, g: _SourceGate) -> float:
+        streak = max(1, g.reopen_streak)
+        return min(self._cfg.cooldown_s * (2 ** (streak - 1)), self._cfg.max_cooldown_s)
+
     def _grant_probe(self, g: _SourceGate, source_name: str, owner: str) -> GateDecision:
         g.state = GateState.HALF_OPEN
         g.probe_owner = owner
@@ -123,18 +169,35 @@ class InMemoryGate:
             else 0.0,
         )
 
-    def _open(self, g: _SourceGate, reason: str) -> None:
+    def _open(self, g: _SourceGate, reason: str, *, bump_streak: bool) -> None:
+        if bump_streak:
+            g.reopen_streak += 1
         g.state = GateState.OPEN
         g.epoch += 1  # 世代推进: 旧 entry 的迟到写回自此被 fencing 拒绝
-        g.open_until = self._now() + self._cfg.cooldown_s
+        g.open_until = self._now() + self._cooldown_eff(g)
         g.fails = max(g.fails, self._cfg.fail_threshold)
         g.probe_owner = None
         g.probe_expires = 0.0
+        g.closed_since = None
 
-    async def record_success(self, entry: GateDecision) -> GateUpdate:
+    async def record_success(
+        self, entry: GateDecision, *, count_attempt: bool = True
+    ) -> GateUpdate:
         g = self._gate(entry.source_name)
         if not self._fenced(g, entry):
             return self._snapshot(g, applied=False)
+        was_probe = entry.is_probe
+        if count_attempt:
+            self._window_add(g, failed=False)
+        # streak 衰减: CLOSED 稳定满 2×cooldown_eff 后的首次成功归零(设计 §3.1)
+        if (
+            g.reopen_streak > 0
+            and g.closed_since is not None
+            and self._now() - g.closed_since >= 2 * self._cooldown_eff(g)
+        ):
+            g.reopen_streak = 0
+        if was_probe:
+            g.closed_since = self._now()  # 仅探针转 CLOSED 时写,普通成功不刷新
         g.state = GateState.CLOSED
         g.fails = 0
         g.probe_owner = None
@@ -147,12 +210,25 @@ class InMemoryGate:
         g = self._gate(entry.source_name)
         if not self._fenced(g, entry):
             return self._snapshot(g, applied=False)
-        if entry.is_probe or force_open:
-            self._open(g, reason)  # 探针失败重开 / SourceDead 一击即熔
+        if reason == "rate_limited" and not force_open:
+            # 429 = 背压不是故障(设计 §3.1): 两通道均不计,选源层软处理;
+            # 探针撞 429 按无果归还语义放下家,不挂租约
+            if entry.is_probe:
+                g.state = GateState.OPEN
+                g.open_until = self._now()
+                g.probe_owner = None
+                g.probe_expires = 0.0
             return self._snapshot(g, applied=True)
+        if entry.is_probe or force_open:
+            # 探针失败重开递增 streak;SourceDead 一击即熔不递增
+            self._open(g, reason, bump_streak=entry.is_probe)
+            return self._snapshot(g, applied=True)
+        self._window_add(g, failed=True)
         g.fails += 1
-        if g.fails >= self._cfg.fail_threshold:
-            self._open(g, reason)
+        if self._rate_channel_open(g):
+            self._open(g, reason, bump_streak=True)
+        elif g.fails >= self._cfg.fail_threshold:
+            self._open(g, reason, bump_streak=False)  # 连续通道不递增(C1 封顶)
         return self._snapshot(g, applied=True)
 
     async def release_probe(self, entry: GateDecision) -> GateUpdate:
diff --git a/src/polygateway/backends/redis/breaker.py b/src/polygateway/backends/redis/breaker.py
index 9e6e5c5..7f772b9 100644
--- a/src/polygateway/backends/redis/breaker.py
+++ b/src/polygateway/backends/redis/breaker.py
@@ -53,9 +53,33 @@ redis.call('HSET', KEYS[1],
 return {1, 'half_open', epoch, 1, ARGV[1], tonumber(ARGV[2])}
 """
 
-# KEYS: provider_hash ; ARGV: expected_epoch is_probe probe_owner(CHS :110-143)
+# M2.5 窗口/退避公共片段(拼接进 success/failure 脚本;Lua 脚本间无法共享函数)
+_WINDOW_HELPERS = """
+local function rotate_window(key, now, half_ms)
+    local wid = math.floor(now / half_ms)
+    local cur = tonumber(redis.call('HGET', key, 'win_id') or '-1')
+    if wid ~= cur then
+        if wid == cur + 1 then
+            redis.call('HSET', key,
+                'a1', redis.call('HGET', key, 'a0') or '0',
+                'f1', redis.call('HGET', key, 'f0') or '0')
+        else
+            redis.call('HSET', key, 'a1', 0, 'f1', 0)
+        end
+        redis.call('HSET', key, 'a0', 0, 'f0', 0, 'win_id', wid)
+    end
+end
+local function cooldown_eff(key, cooldown_ms, max_cooldown_ms)
+    local streak = math.max(1, math.min(
+        tonumber(redis.call('HGET', key, 'reopen_streak') or '0'), 16))
+    return math.floor(math.min(cooldown_ms * 2 ^ (streak - 1), max_cooldown_ms))
+end
+"""
+
+# KEYS: provider_hash(CHS :110-143 + M2.5 窗口/衰减)
+# ARGV: expected_epoch is_probe probe_owner count_attempt half_window_ms cooldown_ms max_cooldown_ms
 # 返回: applied state epoch failures retry_after_ms
-RECORD_SUCCESS = """
+RECORD_SUCCESS = _WINDOW_HELPERS + """
 local t = redis.call('TIME')
 local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
 local state = redis.call('HGET', KEYS[1], 'state') or 'closed'
@@ -71,6 +95,20 @@ else
 end
 
 if matches then
+    if tonumber(ARGV[4]) == 1 then
+        rotate_window(KEYS[1], now, tonumber(ARGV[5]))
+        redis.call('HINCRBY', KEYS[1], 'a0', 1)
+    end
+    -- streak 衰减: CLOSED 稳定满 2×cooldown_eff 后的首次成功归零(设计 §3.1)
+    local streak = tonumber(redis.call('HGET', KEYS[1], 'reopen_streak') or '0')
+    local since = tonumber(redis.call('HGET', KEYS[1], 'closed_since') or '0')
+    if streak > 0 and since > 0
+        and now - since >= 2 * cooldown_eff(KEYS[1], tonumber(ARGV[6]), tonumber(ARGV[7])) then
+        redis.call('HSET', KEYS[1], 'reopen_streak', 0)
+    end
+    if is_probe == 1 then
+        redis.call('HSET', KEYS[1], 'closed_since', now)
+    end
     redis.call('HSET', KEYS[1],
         'state', 'closed',
         'failures', 0,
@@ -90,10 +128,11 @@ end
 return {0, state, epoch, failures, math.max(deadline - now, 0)}
 """
 
-# KEYS: provider_hash(CHS :148-199)
+# KEYS: provider_hash(CHS :148-199 + M2.5 双通道/退避)
 # ARGV: expected_epoch is_probe probe_owner force_open threshold cooldown_ms
+#       rate_limited min_calls fail_rate half_window_ms max_cooldown_ms
 # 返回: applied state epoch failures retry_after_ms
-RECORD_FAILURE = """
+RECORD_FAILURE = _WINDOW_HELPERS + """
 local t = redis.call('TIME')
 local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
 local state = redis.call('HGET', KEYS[1], 'state') or 'closed'
@@ -118,23 +157,60 @@ if not matches then
     return {0, state, epoch, failures, math.max(deadline - now, 0)}
 end
 
+local force_open = tonumber(ARGV[4])
+if tonumber(ARGV[7]) == 1 and force_open == 0 then
+    -- 429 = 背压不是故障(M2.5 设计 §3.1): 两通道均不计;
+    -- 探针撞 429 按无果归还语义放下家,不挂租约
+    if is_probe == 1 then
+        redis.call('HSET', KEYS[1],
+            'state', 'open',
+            'open_until', now,
+            'probe_owner', '',
+            'probe_until', 0)
+        return {1, 'open', epoch, failures, 0}
+    end
+    return {1, state, epoch, failures, 0}
+end
+
 local threshold = tonumber(ARGV[5])
-if tonumber(ARGV[4]) == 1 or is_probe == 1 then
+local open_via_rate = 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)
+    redis.call('HINCRBY', KEYS[1], 'f0', 1)
+    local attempts = tonumber(redis.call('HGET', KEYS[1], 'a0') or '0')
+        + 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
+        open_via_rate = true
+    end
+end
+
+if force_open == 1 or is_probe == 1 then
     failures = threshold
 else
     failures = failures + 1
 end
-if tonumber(ARGV[4]) == 1 or is_probe == 1 or failures >= threshold then
+if force_open == 1 or is_probe == 1 or open_via_rate or failures >= threshold then
+    -- 递增 streak 的只有率通道开路与探针失败重开(C1: 连续通道/force_open 不递增)
+    if open_via_rate or is_probe == 1 then
+        redis.call('HINCRBY', KEYS[1], 'reopen_streak', 1)
+    end
+    if open_via_rate then
+        failures = math.max(failures, threshold)
+    end
     epoch = epoch + 1
-    local open_until = now + tonumber(ARGV[6])
+    local eff = cooldown_eff(KEYS[1], tonumber(ARGV[6]), tonumber(ARGV[11]))
     redis.call('HSET', KEYS[1],
         'state', 'open',
         'failures', failures,
         'epoch', epoch,
-        'open_until', open_until,
+        'open_until', now + eff,
         'probe_owner', '',
         'probe_until', 0)
-    return {1, 'open', epoch, failures, tonumber(ARGV[6])}
+    redis.call('HDEL', KEYS[1], 'closed_since')
+    return {1, 'open', epoch, failures, eff}
 end
 redis.call('HSET', KEYS[1],
     'state', 'closed',
@@ -203,6 +279,10 @@ class RedisGate:
         self._threshold = config.fail_threshold
         self._cooldown_ms = math.ceil(config.cooldown_s * 1000)
         self._probe_ttl_ms = math.ceil(config.probe_ttl_s * 1000)
+        self._max_cooldown_ms = math.ceil(config.max_cooldown_s * 1000)
+        self._half_window_ms = max(1, math.ceil(config.window_s * 500))  # 半窗
+        self._min_calls = config.min_calls
+        self._fail_rate = config.fail_rate
         self._redis = redis
         self._owns_client = False
         self._try_enter_lua = redis.register_script(TRY_ENTER)
@@ -277,11 +357,15 @@ class RedisGate:
             raise GovernanceBackendError(f"熔断后端 try_enter 失败: {exc}") from exc
         return self._decision(source_name, result)
 
-    async def record_success(self, entry: GateDecision) -> GateUpdate:
+    async def record_success(
+        self, entry: GateDecision, *, count_attempt: bool = True
+    ) -> GateUpdate:
+        args = self._entry_args(entry)
+        args.extend(
+            [1 if count_attempt else 0, self._half_window_ms, self._cooldown_ms, self._max_cooldown_ms]
+        )
         try:
-            result = await self._success_lua(
-                keys=[self._key(entry.source_name)], args=self._entry_args(entry)
-            )
+            result = await self._success_lua(keys=[self._key(entry.source_name)], args=args)
         except RedisError as exc:
             raise GovernanceBackendError(f"熔断后端 record_success 失败: {exc}") from exc
         return self._update(result)
@@ -290,7 +374,18 @@ class RedisGate:
         self, entry: GateDecision, reason: str, force_open: bool
     ) -> GateUpdate:
         args = self._entry_args(entry)
-        args.extend([1 if force_open else 0, self._threshold, self._cooldown_ms])
+        args.extend(
+            [
+                1 if force_open else 0,
+                self._threshold,
+                self._cooldown_ms,
+                1 if reason == "rate_limited" else 0,
+                self._min_calls,
+                self._fail_rate,
+                self._half_window_ms,
+                self._max_cooldown_ms,
+            ]
+        )
         try:
             result = await self._failure_lua(keys=[self._key(entry.source_name)], args=args)
         except RedisError as exc:
diff --git a/tests/contracts/test_breaker_contract.py b/tests/contracts/test_breaker_contract.py
index c9624d9..2ffabea 100644
--- a/tests/contracts/test_breaker_contract.py
+++ b/tests/contracts/test_breaker_contract.py
@@ -133,6 +133,147 @@ class TestEpochFencing:
         assert not (await gate.record_success(stale_probe)).applied
 
 
+_RATE_CFG = BreakerConfig(
+    fail_threshold=100,  # 连续通道抬高失声,单测率通道
+    cooldown_s=60.0,
+    probe_ttl_s=120.0,
+    min_calls=4,
+    fail_rate=0.9,
+    window_s=60.0,
+    max_cooldown_s=240.0,
+)
+
+
+async def _fail(gate, source="s1", reason="timeout", n=1):
+    update = None
+    for _ in range(n):
+        entry = await gate.try_enter(source, "w")
+        assert entry.allowed
+        update = await gate.record_failure(entry, reason, False)
+    return update
+
+
+class TestRateChannel:
+    """M2.5 失败率通道(设计 §3.1;全部确定性序列)。"""
+
+    async def test_rate_opens_at_min_calls(self, gate_factory):
+        # 病灶 1 回归: 高失败率源在 min_calls 样本处开路(连续通道静默)
+        gate = gate_factory(_RATE_CFG)
+        await _fail(gate, n=3)  # attempts 3 < min_calls 4 → 仍 CLOSED
+        assert (await gate.try_enter("s1", "w")).allowed
+        update = await _fail(gate, n=1)  # attempts 4, 失败率 1.0 ≥ 0.9
+        assert update.state is GateState.OPEN
+        assert not (await gate.try_enter("s1", "w")).allowed
+
+    async def test_below_min_calls_never_rate_opens(self, gate_factory):
+        gate = gate_factory(_RATE_CFG)
+        await _fail(gate, n=3)
+        assert (await gate.try_enter("s1", "w")).allowed
+
+    async def test_real_success_dilutes_window(self, gate_factory):
+        # 真实成功计入 attempts: 3 失败 + 1 成功 + 1 失败 = 4/5 = 0.8 < 0.9 不开
+        gate = gate_factory(_RATE_CFG)
+        await _fail(gate, n=3)
+        entry = await gate.try_enter("s1", "w")
+        await gate.record_success(entry)
+        update = await _fail(gate, n=1)
+        assert update.state is GateState.CLOSED
+        assert (await gate.try_enter("s1", "w")).allowed
+
+    async def test_result_invalid_not_counted(self, gate_factory):
+        # 坏结果 ≠ 坏服务: count_attempt=False 完全不动窗口 →
+        # 3 失败 + 1 不计成功 + 1 失败 = attempts 4, 失败率 1.0 → 开路
+        gate = gate_factory(_RATE_CFG)
+        await _fail(gate, n=3)
+        entry = await gate.try_enter("s1", "w")
+        await gate.record_success(entry, count_attempt=False)
+        update = await _fail(gate, n=1)
+        assert update.state is GateState.OPEN
+
+    async def test_rate_limited_bypasses_both_channels(self, gate_factory):
+        # C1 对抗: 429 是背压不是故障——连续通道(阈值 3)与率通道都不吃
+        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=4)
+        gate = gate_factory(cfg)
+        await _fail(gate, reason="rate_limited", n=10)
+        assert (await gate.try_enter("s1", "w")).allowed
+        # timeout 脉冲照常走连续通道开路
+        update = await _fail(gate, reason="timeout", n=3)
+        assert update.state is GateState.OPEN
+
+    async def test_probe_rate_limited_releases_not_hangs(self, gate_factory, clock):
+        # 探针撞 429: 非故障证据也非成功——按无果归还语义放下家接管,
+        # 不得把探针租约挂到 TTL(否则源被锁死 probe_ttl_s)
+        gate = gate_factory(_RATE_CFG)
+        await _fail(gate, n=4)
+        clock.advance(61)
+        probe = await gate.try_enter("s1", "w1")
+        assert probe.is_probe
+        update = await gate.record_failure(probe, "rate_limited", False)
+        assert update.applied
+        nxt = await gate.try_enter("s1", "w2")
+        assert nxt.allowed and nxt.is_probe  # 立即可再探,而非等 probe_ttl
+
+    async def test_rate_open_failure_count_capped(self, gate_factory):
+        gate = gate_factory(_RATE_CFG)
+        await _fail(gate, n=3)
+        update = await _fail(gate, n=1)
+        assert update.state is GateState.OPEN
+        assert update.failure_count == _RATE_CFG.fail_threshold  # 顶格语义沿用
+
+
+class TestReopenBackoff:
+    """开路时长指数递增与衰减(设计 §3.1;memory 假时钟,redis 走真实等待变体)。"""
+
+    async def test_backoff_doubles_and_caps(self, gate_factory, clock):
+        gate = gate_factory(_RATE_CFG)
+        await _fail(gate, n=4)  # 率通道首开: cooldown_eff = 60
+        assert 0 < await gate.retry_after_s(("s1",)) <= 60.0
+        clock.advance(61)
+        probe = await gate.try_enter("s1", "w1")
+        assert probe.is_probe
+        await gate.record_failure(probe, "timeout", False)  # 探针失败重开: streak 2 → 120
+        wait = await gate.retry_after_s(("s1",))
+        assert 60.0 < wait <= 120.0
+        clock.advance(121)
+        probe = await gate.try_enter("s1", "w1")
+        await gate.record_failure(probe, "timeout", False)  # streak 3 → 240(封顶)
+        wait = await gate.retry_after_s(("s1",))
+        assert 120.0 < wait <= 240.0
+
+    async def test_streak_survives_close_then_decays(self, gate_factory, clock):
+        gate = gate_factory(_RATE_CFG)
+        await _fail(gate, n=4)
+        clock.advance(61)
+        probe = await gate.try_enter("s1", "w1")
+        await gate.record_failure(probe, "timeout", False)  # streak 2
+        clock.advance(121)
+        probe = await gate.try_enter("s1", "w1")
+        await gate.record_success(probe)  # 转 CLOSED,streak 不清零
+        # 窗口含探针成功 1 次,9 失败 → 9/10 = 0.9 率开;streak 递增至封顶
+        await _fail(gate, n=9)
+        assert await gate.retry_after_s(("s1",)) > 60.0
+        # 衰减: CLOSED 稳定 2×cooldown_eff 后首次 record_success 归零
+        clock.advance(241)
+        probe = await gate.try_enter("s1", "w1")
+        await gate.record_success(probe)
+        clock.advance(2 * 240.0 + 1)
+        entry = await gate.try_enter("s1", "w")
+        await gate.record_success(entry)  # 触发衰减
+        await _fail(gate, n=9)  # 再开(9/10)回到基础档
+        assert 0 < await gate.retry_after_s(("s1",)) <= 60.0
+
+    async def test_consecutive_open_does_not_bump_streak(self, gate_factory, clock):
+        # C1 对抗: 连续通道误熔健康源的代价封顶为单次 cooldown_s
+        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
+        gate = gate_factory(cfg)
+        await _fail(gate, reason="timeout", n=3)  # 连续通道开路
+        clock.advance(61)
+        probe = await gate.try_enter("s1", "w1")
+        await gate.record_success(probe)  # 恢复
+        await _fail(gate, reason="timeout", n=3)  # 再次连续开路
+        assert 0 < await gate.retry_after_s(("s1",)) <= 60.0  # 无翻倍
+
+
 class TestRetryAfter:
     async def test_retry_after_semantics(self, gate_factory, clock):
         gate = gate_factory(_CFG)
diff --git a/tests/integration/test_redis_governance_time.py b/tests/integration/test_redis_governance_time.py
index dd14471..cf26460 100644
--- a/tests/integration/test_redis_governance_time.py
+++ b/tests/integration/test_redis_governance_time.py
@@ -235,3 +235,95 @@ async def test_rpm_window_rollover_resets_quota(redis_client):
     fresh = await limiter.try_acquire("s1", 0)
     assert fresh is not None
     await fresh.release()
+
+
+# —— M2.5 双通道/退避变体(小 cooldown 配置控时长;语义与契约用例同源)——
+
+_M25_CFG = BreakerConfig(
+    fail_threshold=100,
+    cooldown_s=4.0,
+    probe_ttl_s=8.0,
+    min_calls=4,
+    fail_rate=0.9,
+    window_s=8.0,
+    max_cooldown_s=16.0,
+)
+_M25_CONSEC_CFG = BreakerConfig(fail_threshold=3, cooldown_s=4.0, probe_ttl_s=8.0)
+
+
+def _m25_gate(redis_client, cfg=_M25_CFG) -> RedisGate:
+    return RedisGate(config=cfg, redis=redis_client, scope=f"t{uuid4().hex[:8]}")
+
+
+async def _fail_n(gate, n, reason="timeout"):
+    update = None
+    for _ in range(n):
+        entry = await gate.try_enter("s1", "w")
+        assert entry.allowed
+        update = await gate.record_failure(entry, reason, False)
+    return update
+
+
+@pytestmark_slow
+async def test_variant_backoff_doubles_and_caps(redis_client):
+    gate = _m25_gate(redis_client)
+    update = await _fail_n(gate, 4)  # 率通道首开(4/4 ≥ 0.9): cooldown_eff = 4
+    assert update.state is GateState.OPEN
+    assert 0 < await gate.retry_after_s(("s1",)) <= 4.0
+    await asyncio.sleep(5)
+    probe = await gate.try_enter("s1", "w1")
+    assert probe.is_probe
+    await gate.record_failure(probe, "timeout", False)  # streak 2 → 8
+    wait = await gate.retry_after_s(("s1",))
+    assert 4.0 < wait <= 8.0
+    await asyncio.sleep(9)
+    probe = await gate.try_enter("s1", "w1")
+    await gate.record_failure(probe, "timeout", False)  # streak 3 → 16(封顶)
+    wait = await gate.retry_after_s(("s1",))
+    assert 8.0 < wait <= 16.0
+
+
+@pytestmark_slow
+async def test_variant_streak_survives_close_then_decays(redis_client):
+    gate = _m25_gate(redis_client)
+    await _fail_n(gate, 4)  # streak 1(4s)
+    await asyncio.sleep(5)
+    probe = await gate.try_enter("s1", "w1")
+    await gate.record_failure(probe, "timeout", False)  # streak 2(8s)
+    await asyncio.sleep(9)
+    probe = await gate.try_enter("s1", "w1")
+    await gate.record_success(probe)  # 转 CLOSED,streak 不清零
+    await _fail_n(gate, 9)  # 窗口 1 成功 + 9 失败 = 0.9 率开: streak 3 → 16
+    assert await gate.retry_after_s(("s1",)) > 8.0
+    await asyncio.sleep(17)
+    probe = await gate.try_enter("s1", "w1")
+    await gate.record_success(probe)  # closed_since 落点
+    await asyncio.sleep(2 * 16.0 + 1)
+    entry = await gate.try_enter("s1", "w")
+    await gate.record_success(entry)  # CLOSED 稳定期满 → streak 衰减归零
+    await _fail_n(gate, 9)  # 再开回基础档
+    assert 0 < await gate.retry_after_s(("s1",)) <= 4.0
+
+
+@pytestmark_slow
+async def test_variant_consecutive_open_does_not_bump_streak(redis_client):
+    gate = _m25_gate(redis_client, _M25_CONSEC_CFG)
+    await _fail_n(gate, 3)  # 连续通道开路(不递增 streak)
+    await asyncio.sleep(5)
+    probe = await gate.try_enter("s1", "w1")
+    await gate.record_success(probe)
+    await _fail_n(gate, 3)  # 再次连续开路: 仍是基础档
+    assert 0 < await gate.retry_after_s(("s1",)) <= 4.0
+
+
+@pytestmark_slow
+async def test_variant_probe_rate_limited_releases_not_hangs(redis_client):
+    gate = _m25_gate(redis_client)
+    await _fail_n(gate, 4)
+    await asyncio.sleep(5)
+    probe = await gate.try_enter("s1", "w1")
+    assert probe.is_probe
+    update = await gate.record_failure(probe, "rate_limited", False)
+    assert update.applied
+    nxt = await gate.try_enter("s1", "w2")
+    assert nxt.allowed and nxt.is_probe  # 立即可再探,不等 probe_ttl