feat: let circuit_open=wait queue instead of killing the call

on_no_runnable now dispatches on why every source was rejected instead
of falling through two serial branches. Under wait, a fully open circuit
sleeps out the cooldown and comes back for another round; the breaker's
protection is untouched (still not a single request leaves during the
wait, so no quota or money burns) -- what changes is whether the caller
dies on the spot or queues.

Dispatching is not cosmetic. Left serial, wait would fall into the quota
branch and a caller with quota_full=fail_fast would get a
quota_exhausted error while its quota was in fact fine.

_nap sleeps to the cooldown deadline rather than polling every 10ms,
which for a 60s cooldown is 6000 round trips per in-flight call on the
Redis backend. Jitter is added on top instead of scaling the wait, since
waking early before a known deadline just earns another rejection. Both
arms clamp to the remaining stall budget, so the worst case per call is
stall_window plus one poll and does not drift with max_cooldown_s. The
clamp's lower bound is the jitter itself, not poll_interval -- the
latter would have lifted the existing [0.5p, 1.0p] quota polling.

src/polygateway/middleware/admission.py

@@ -224,14 +224,26 @@ class SourceAdmission:
     async def on_no_runnable(
         self, gate_rejections: int, reasons: dict[str, str], clock: StallClock
     ) -> None:
-        """一个源都挑不出来时的处置: 判死或等待一轮再来。"""
+        """一个源都挑不出来时的处置: **按拒绝原因分派**到各自的策略。
-        if gate_rejections == len(self._sources):
+
+        分派而非串行是硬要求(issue #14): 串行写法下 `circuit_open=wait` 不抛
+        之后会径直掉进配额分支,`quota_full=fail_fast` 的调用方于是收到一个
+        `reason=quota_exhausted` 的异常——而配额其实是满的,坏的是熔断门。
+        """
         names = tuple(s.name for s in self._sources)
+        if gate_rejections == len(self._sources):
+            # 全部因熔断类原因(门开路 / 本地冷却备忘)被拒
+            if self._circuit_open == "fail_fast":
                 raise CircuitOpenError(
                     scope=self._scope,
                     retry_after_s=await self._breaker.retry_after_s(names),
                     per_source_reasons=reasons,
                 )
+            # wait: 保护作用完整保留(这一轮照样一个请求都不发),改变的只是
+            # 调用方当场死还是排队等——多源可换源故 fail-fast 对,单源无源可换
+            hint = await self._breaker.retry_after_s(names)
+        else:
+            # 至少一个源是被配额/AIMD 挡的,归 quota_full 管
             if self._quota_full == "fail_fast":
                 raise AllSourcesExhausted(
                     scope=self._scope,
@@ -239,13 +251,39 @@ class SourceAdmission:
                     retry_after_s=self._bp.poll_interval_s,
                     per_source_reasons=reasons,
                 )
+            hint = 0.0
         if await self.stalled(clock):
-            names = tuple(s.name for s in self._sources)
             raise AllSourcesExhausted(
                 scope=self._scope,
                 reason="stalled",
                 retry_after_s=await self._breaker.retry_after_s(names),
                 per_source_reasons=reasons,
             )
-        # jitter ∈ [0.5p, 1.0p] 防惊群(CHS governance.py:283-285)
+        nap = self._nap(hint, clock)
-        await self._sleep(self._bp.poll_interval_s * (0.5 + 0.5 * self._rng()))
+        if hint > 0:
+            logger.info(
+                "熔断开路等待 {:.1f}s 后重试(scope={}, 原因={})", nap, self._scope, reasons
+            )
+        await self._sleep(nap)
+
+    def _nap(self, hint: float, clock: StallClock) -> float:
+        """本轮等待多久。**必须在 `stalled()` 判定之后调用**(预算可能已耗尽)。
+
+        `hint > 0`(熔断开路有确定的冷却截止)时睡到那个时刻,而不是按
+        `poll_interval` 空转——60 秒冷却用 10ms 轮询是 6000 次空转,内存后端
+        只是查字典,Redis 后端则是 6000 次往返 × 每个在途调用。抖动**上**加
+        而非缩放(既有 quota 路径是 `[0.5p, 1.0p]`): 对一个确定的截止时刻提前
+        醒来必然被再拒一次,白跑一趟。
+
+        两档都夹到剩余 stall 预算,故单次调用的最坏墙钟是 `stall_window_s`
+        加一个 poll 间隔,不随 `max_cooldown_s` 漂移。多加的那一格是因为
+        `stalled()` 判据是 `>` 而非 `>=`——恰好睡到窗口边界不判死,留这一格
+        让下一轮必定判死。`hint == 0` 时整个式子退化为既有的 jitter 轮询。
+        """
+        jitter = self._bp.poll_interval_s * (0.5 + 0.5 * self._rng())
+        budget = self._bp.stall_window_s - clock.stalled_s() + self._bp.poll_interval_s
+        wait = hint + jitter if hint > 0 else jitter
+        # 下界取 jitter 而非 poll_interval: 既有 quota 轮询是 [0.5p, 1.0p],用
+        # poll_interval 兜底会把 rng→0 那半边抬上去。预算为负时(本地已超窗但
+        # 全局仍在出餐,故 stalled() 不判死)靠它退回正常轮询节奏,不忙循环。
+        return max(jitter, min(wait, budget))

tests/unit/test_backpressure.py

@@ -13,6 +13,7 @@ from polygateway.backends.memory.breaker import InMemoryGate
 from polygateway.backends.memory.limiter import InMemoryLimiter
 from polygateway.errors import (
     AllSourcesExhausted,
+    CircuitOpenError,
     GatewayUnavailableError,
     GovernanceBackendError,
     SourceNotConfiguredError,
@@ -62,6 +63,7 @@ def _mw(
     sleep,
     rng=lambda: 0.0,
     quota_full="wait",
+    circuit_open="fail_fast",
     gate=None,
     transport=None,
     emitter=None,
@@ -76,6 +78,7 @@ def _mw(
         retry=RetryPolicy(max_attempts=3, backoff_base_s=2.0, backoff_max_s=30.0),
         backpressure=BackpressurePolicy(stall_window_s=_STALL, poll_interval_s=0.01),
         quota_full=quota_full,
+        circuit_open=circuit_open,
         cooldown_memo=SourceCooldownMemo(now=clock),
         emitter=emitter,
         now=clock,
@@ -593,3 +596,143 @@ class TestGateFailuresReachCallersAsScopeLevel:
             await QuotaGate(_Broken(), scope="LLM").progress_age_s()
         assert ei.value.scope == "llm"
         assert ei.value.reason == "governance_backend_down"
+
+
+class TestCircuitOpenPolicy:
+    """issue #14: 熔断全拒时是当场判死还是等冷却过去。
+
+    缺省 fail_fast 即历史行为(TestStallQuadrants 等既有用例照旧覆盖);
+    本类钉的是 wait 档,以及两条策略互不串线。
+    """
+
+    @staticmethod
+    async def _opened_gate(clock, cfg=_BREAKER):
+        gate = InMemoryGate(config=cfg, now=clock)
+        for _ in range(cfg.fail_threshold):
+            entry = await gate.try_enter("s1", "w")
+            await gate.record_failure(entry, "network_error", False)
+        return gate
+
+    @staticmethod
+    def _free_limiter(clock, src):
+        return InMemoryLimiter(
+            scope="llm",
+            sources={"s1": src},
+            global_limits=_NO_GLOBAL,
+            lease_ttl_s=10_000.0,
+            now=clock,
+        )
+
+    async def test_fail_fast_is_the_default(self):
+        """缺省档逐字保持历史行为: 全源开路当场抛 CircuitOpenError。"""
+        clock = FakeClock()
+        src = make_source()
+        mw = _mw(
+            [src], self._free_limiter(clock, src), [], clock=clock,
+            sleep=BoundedSleep(), gate=await self._opened_gate(clock),
+        )
+        with pytest.raises(CircuitOpenError) as ei:
+            await mw(_REQ)
+        assert ei.value.reason == "circuit_open"
+
+    async def test_wait_sleeps_out_the_cooldown_instead_of_dying(self):
+        """wait 档: 睡到冷却结束再来一轮,拿到探针后正常返回。
+
+        睡的是**冷却剩余**而不是 poll_interval——60 秒冷却用 10ms 轮询要空转
+        6000 次,memory 后端只是查字典,Redis 后端则是 6000 次往返 × 每个在途调用。
+        """
+        clock = FakeClock()
+        src = make_source()
+        sleep = BoundedSleep()
+
+        async def advance(_n):
+            clock.advance(sleep.delays[-1])
+
+        sleep._side_effect = advance
+        mw = _mw(
+            [src], self._free_limiter(clock, src), [_ok()], clock=clock,
+            sleep=sleep, gate=await self._opened_gate(clock), circuit_open="wait",
+        )
+        resp = await mw(_REQ)
+        assert resp.content == "ok"
+        # 一觉睡到冷却结束(jitter 上加,rng=0 → +0.5×poll),不是 poll 空转
+        assert sleep.delays[0] == pytest.approx(_BREAKER.cooldown_s + 0.005)
+
+    async def test_wait_does_not_leak_into_the_quota_branch(self):
+        """两条策略互不串线: circuit_open=wait 配 quota_full=fail_fast 时,
+        熔断等待**不得**被当成配额耗尽上报——串线会让调用方拿到一个
+        reason=quota_exhausted 的异常,而配额其实是满的。"""
+        clock = FakeClock()
+        src = make_source()
+        sleep = BoundedSleep()
+
+        async def advance(_n):
+            clock.advance(sleep.delays[-1])
+
+        sleep._side_effect = advance
+        mw = _mw(
+            [src], self._free_limiter(clock, src), [_ok()], clock=clock, sleep=sleep,
+            gate=await self._opened_gate(clock), quota_full="fail_fast", circuit_open="wait",
+        )
+        assert (await mw(_REQ)).content == "ok"
+
+    async def test_wait_still_dies_when_cooldown_outlasts_the_stall_budget(self):
+        """等待有可解释的上界: 冷却比 stall 预算还长时,在窗口耗尽处判死。
+
+        单次睡眠夹到剩余 stall 预算,故最坏墙钟 = stall_window + 一个 poll,
+        不随 max_cooldown_s 漂移。
+        """
+        clock = FakeClock()
+        src = make_source()
+        long_cooldown = BreakerConfig(
+            fail_threshold=3, cooldown_s=1000.0, probe_ttl_s=2000.0, max_cooldown_s=1000.0
+        )
+        sleep = BoundedSleep()
+
+        async def advance(_n):
+            clock.advance(sleep.delays[-1])
+
+        sleep._side_effect = advance
+        mw = _mw(
+            [src], self._free_limiter(clock, src), [], clock=clock, sleep=sleep,
+            gate=await self._opened_gate(clock, long_cooldown), circuit_open="wait",
+        )
+        with pytest.raises(AllSourcesExhausted) as ei:
+            await mw(_REQ)
+        assert ei.value.reason == "stalled"
+        assert ei.value.per_source_reasons == {"s1": "circuit_open"}
+        assert sleep.delays[0] == pytest.approx(_STALL + 0.01)  # 夹到预算 + 一个 poll
+
+    async def test_wait_loop_stays_cancellable(self):
+        """取消穿透(铁律): 熔断等待中的取消不得被吞。"""
+        clock = FakeClock()
+        src = make_source()
+        mw = _mw(
+            [src], self._free_limiter(clock, src), [], clock=clock,
+            sleep=asyncio.sleep, gate=await self._opened_gate(clock), circuit_open="wait",
+        )
+        task = asyncio.create_task(mw(_REQ))
+        await asyncio.sleep(0.03)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    async def test_half_open_rejection_does_not_blacklist_a_recovered_source(self):
+        """issue #14 §1.3 回归: 探针成功后本进程立即可再选该源。
+
+        此前 HALF_OPEN 拒绝把探针租约(派生自 2 × timeout,现场 600s)写进冷却
+        备忘,而 `set_until` 取更晚者、不可回退——门恢复 CLOSED 之后本进程仍
+        跳过该源整整一个租约,单源下每次调用照旧判死。多源部署同样中招,只是
+        被别的源接住流量掩盖了。
+        """
+        clock = FakeClock()
+        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=600.0)
+        gate = await self._opened_gate(clock, cfg)
+        memo = SourceCooldownMemo(now=clock)
+        clock.advance(cfg.cooldown_s + 1)
+        probe = await gate.try_enter("s1", "w1")
+        blocked = await gate.try_enter("s1", "w2")  # 并发调用撞上在途探针
+        assert not blocked.allowed
+        memo.set_until("s1", clock() + blocked.retry_after_s)  # 准入路径的写法
+        await gate.record_success(probe)  # 探针成功 → 门恢复 CLOSED
+        assert not memo.active("s1")