fix: skip hedge dispatch when primary finishes during admission

Address branch review findings 1-4 on feature/1.3.7-hedged-requests:

- src/polygateway/middleware/retry.py: recheck primary.done() after
  hedge admission in _attempt_hedged; release the hedge permit via
  settle_and_release(permit, 0) (release_probe for probe entries) and
  adjudicate the primary directly instead of firing a billable HTTP
  request that would be cancelled immediately
- src/polygateway/config.py: check_hedge_assembly raises a hedge-located
  ValueError for empty sources instead of a bare min() error
- tests/unit/test_hedge.py: pin that an injected FakeClock jump of 10^6
  seconds does not trigger hedging (design section 8); pin pick(exclude)
  counting no gate_rejections and leaving reasons untouched; pin silent
  hedge abandonment when the candidate circuit is open; deterministic
  regression for the admission-window race (BlockingLimiter harness)
- tests/unit/test_config.py: assert the empty-sources guard message
  locates the hedge key

Red-to-green evidence in tests/outputs/137/review-fixes/

src/polygateway/config.py

@@ -809,6 +809,13 @@ def check_hedge_assembly(
     after = ensure_call_deadline(hedge_after_s, origin)
     if after is None:
         return None
+    if not sources:
+        # 直传路(GatewayClient(sources=[], hedge_after_s=...))没有 settings 的
+        # 非空守卫先行拦截;报错必须定位到 hedge,而不是裸 min() 空序列异常
+        raise ValueError(
+            "hedge_after_s({SCOPE}__HEDGE__AFTER_S)的交叉守卫要求 sources 不能为空;"
+            "对冲已启用但没有可校验的源"
+        )
     min_timeout = min(s.timeout_s for s in sources)
     if after >= min_timeout:
         raise ValueError(

src/polygateway/middleware/retry.py

@@ -456,6 +456,22 @@ class RetryMW:
                 if isinstance(outcome, LLMResponse):
                     self._record_generation(request, sink_p)
                 return outcome
+            # Phase 3.5 准入后复查: 原路可能在对冲准入的 await 期间已了结——此时
+            # 一个对冲请求都不发(那是白付一次真实计费请求 + 一份 est 滞留 + 一条
+            # hedge_cancelled 行)。按既有语义释放刚拿到的对冲准入(probe 须
+            # release_probe,顺序同 _attempt 取消分支),直接裁定原路结果
+            if primary.done():
+                hedge_source, hedge_permit, hedge_entry = hedge_picked
+                try:
+                    if hedge_entry.is_probe:
+                        await self._record_quietly(self._breaker.release_probe(hedge_entry))
+                finally:
+                    self._pacer.leave(hedge_source.name)
+                    await settle_and_release(hedge_permit, 0)
+                outcome = await primary
+                if isinstance(outcome, LLMResponse):
+                    self._record_generation(request, sink_p)
+                return outcome
             # Phase 4 启动对冲路(v1 单路,H5): 对冲路恒传 first_token_event=None,
             # 不再触发梯次对冲
             sink_h: list[int] = []

tests/unit/test_config.py

@@ -1166,6 +1166,31 @@ class TestHedgeConfig:
         with pytest.raises(ValueError, match=r"GatewayClient\(hedge_after_s"):
             _client(hedge_after_s=0)
 
+    def test_hedge_guard_empty_sources_raises_with_hedge_location(self):
+        """直传路空 sources + 设阈值: ValueError 且消息定位到 hedge(不是裸 min() 报错)。"""
+        from polygateway.config import check_hedge_assembly
+
+        with pytest.raises(ValueError, match="sources") as ei:
+            check_hedge_assembly(
+                hedge_after_s=8,
+                hedge_max_extra=1,
+                sources=[],
+                call_deadline_s=None,
+                origin="GatewayClient(hedge_after_s=8)",
+            )
+        assert "hedge_after_s" in str(ei.value)  # 定位得到是哪个键
+        # 未启用对冲(None)时空 sources 直接放行: 交叉守卫没有可校验的对象
+        assert (
+            check_hedge_assembly(
+                hedge_after_s=None,
+                hedge_max_extra=1,
+                sources=[],
+                call_deadline_s=None,
+                origin="test",
+            )
+            is None
+        )
+
     def test_hedge_guard_below_min_timeout_raises(self):
         """阈值 ≥ 最小源 timeout_s = 对冲永不可能触发,装配期炸掉(ValueError)。"""
         env = self._two_source_env(**{"LLM__HEDGE__AFTER_S": "90"})  # min(timeout)=90

tests/unit/test_hedge.py

@@ -25,6 +25,7 @@ from polygateway.types import (
     RetryPolicy,
     _CallContext,
 )
+from tests.contracts.conftest import FakeClock
 from tests.unit.test_retry import RecordingSelector, StaticSelector, _ok, _src
 
 _BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
@@ -80,6 +81,38 @@ class HedgeTransport:
         raise AssertionError(f"未知剧本动作: {action!r}")
 
 
+class BlockingLimiter:
+    """限流包装: 挂起指定源的 try_acquire 直到测试放行(确定性复现"对冲准入挂起")。
+
+    `acquiring` 置位 = 对冲准入已停在该源闸内;`allow` 置位后才继续。只拦对冲
+    会走到的源,原路准入不受影响;其余方法逐字委托内层 InMemoryLimiter。
+    """
+
+    def __init__(self, inner: InMemoryLimiter, block_source: str):
+        self._inner = inner
+        self._block_source = block_source
+        self.acquiring = asyncio.Event()
+        self.allow = asyncio.Event()
+
+    async def try_acquire(self, source_key, est_tokens):
+        if source_key == self._block_source:
+            self.acquiring.set()
+            await self.allow.wait()
+        return await self._inner.try_acquire(source_key, est_tokens)
+
+    async def acquire(self, source_key, est_tokens):
+        return await self._inner.acquire(source_key, est_tokens)
+
+    async def source_stats(self, source_key):
+        return await self._inner.source_stats(source_key)
+
+    async def mark_progress(self):
+        return await self._inner.mark_progress()
+
+    async def progress_age_s(self):
+        return await self._inner.progress_age_s()
+
+
 class RecordingEmitter:
     """逐次遥测假 emitter: 记录每行的源/错误标签/逻辑调用 ID/attempt call_id。"""
 
@@ -121,8 +154,13 @@ def _harness(
     limiter=None,
     gate=None,
     stall_window_s=300.0,
+    now=None,
 ):
-    """真实 loop 钟装配(对冲计时纪律: 只用 loop 相对时长,不注入 FakeClock)。"""
+    """真实 loop 钟装配(对冲计时纪律: 只用 loop 相对时长,不注入 FakeClock)。
+
+    `now` 仅供"注入钟与对冲触发正交"用例注入 FakeClock——触发路径结构性不读
+    它,注入只是为了证明这一点。
+    """
     limiter = limiter or InMemoryLimiter(
         scope="llm",
         sources={s.name: s for s in sources},
@@ -144,6 +182,7 @@ def _harness(
         cooldown_memo=SourceCooldownMemo(),
         emitter=emitter,
         hedge_after_s=hedge_after_s,
+        **({"now": now} if now is not None else {}),
     )
     return mw, limiter, gate
 
@@ -209,6 +248,30 @@ class TestHedgeTrigger:
         stats2 = ctx2.snapshot()
         assert stats2.hedges == 0 and stats2.hedge_won is False and stats2.attempts == 1
 
+    async def test_injected_clock_jump_does_not_trigger_hedge(self):
+        """对冲触发只认真实 loop 钟: 注入钟跳 10^6 秒不得触发对冲(设计 §8 验收矩阵)。
+
+        s1 挂起剧本 + 触发窗内注入钟拨快 10^6 秒: 若触发路径误读注入钟,对冲会
+        **立即**发出;断言对冲实际发出时刻不早于真实 loop 阈值(下界断言,不断
+        精确值),形态同 test_client.py:1974 deadline 的注入钟对应用例。
+        """
+        clock = FakeClock()
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "hedged")]})
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, now=clock)
+        ctx = _ctx()
+        task = asyncio.ensure_future(mw(_req(ctx=ctx)))
+        await transport.entered["s1"].wait()  # 原路在途,触发窗计时中
+        clock.advance(1_000_000.0)  # 跳变落在窗内: 误读注入钟即立刻触发
+        started = time.monotonic()
+        resp = await asyncio.wait_for(task, timeout=5)
+        elapsed = time.monotonic() - started
+        assert resp.source_name == "s2"  # 对冲确由真实 loop 阈值触发并截断长尾
+        assert transport.calls == ["s1", "s2"]
+        # 下界留 20% 调度余量;误读注入钟的触发是毫秒级,与此差一个数量级以上
+        assert elapsed >= _HEDGE_AFTER_S * 0.8
+        stats = ctx.snapshot()
+        assert stats.hedges == 1 and stats.hedge_won is True
+
 
 class TestHedgeRouting:
     """异源排除与静默放弃(验收矩阵 ②③)。"""
@@ -254,6 +317,81 @@ class TestHedgeRouting:
         finally:
             await held.release()
 
+    async def test_pick_exclude_all_is_not_a_rejection(self):
+        """exclude 覆盖全源 → 返回 None 且 gate_rejections==0、reasons 不写(排除 ≠ 拒绝)。
+
+        admission 级直接钉(admission.py:192 `continue` 语义): 若未来重构把排除计入
+        gate_rejections,`on_no_runnable` 的"全源熔断类拒绝"判据会被污染,此钉当场报警。
+        """
+        transport = HedgeTransport({"s1": [("succeed", "x")], "s2": [("succeed", "y")]})
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport)
+        reasons = {"prior": "rate_limited"}  # 既有原因须原样保留
+        picked, gate_rejections = await mw._admission.pick(
+            reasons, {}, exclude=frozenset({"s1", "s2"})
+        )
+        assert picked is None
+        assert gate_rejections == 0
+        assert reasons == {"prior": "rate_limited"}
+
+    async def test_hedge_silent_when_candidate_circuit_open(self):
+        """对冲候选被熔断开路 → 静默放弃: 不对冲、不抛错、原请求照等(②③的另一形态)。
+
+        现有限流闸用例只钉了"配额占满"一条静默路径;开路/pacer 拒绝走 pick 的另一
+        分支(gate_rejections 计数、reasons 写 circuit_open、settle_and_release 后
+        返回 None),同样不得发出对冲请求。
+        """
+        gate = InMemoryGate(config=_BREAKER)
+        entry = await gate.try_enter("s2", "test-owner")
+        assert entry.allowed
+        await gate.record_failure(entry, "source_dead", True)  # SourceDead 一击即熔
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "x")]})
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, gate=gate)
+        ctx = _ctx()
+        task = asyncio.ensure_future(mw(_req(ctx=ctx)))
+        await transport.entered["s1"].wait()
+        # 4× 余量: 给对冲窗与那次注定被开路拒绝的准入留足发生时间
+        await asyncio.sleep(4 * _HEDGE_AFTER_S)
+        assert transport.calls == ["s1"]  # 对冲静默未发出
+        transport.release["s1"].set()
+        resp = await asyncio.wait_for(task, timeout=5)
+        assert resp.source_name == "s1"
+        stats = ctx.snapshot()
+        assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False
+
+    async def test_primary_done_during_hedge_admission_sends_no_hedge(self):
+        """原路在对冲准入 await 期间已完成: 释放对冲准入直接裁定,一个对冲请求都不发。
+
+        剧本钉死窗口(禁 sleep 猜): s2 的 try_acquire 挂起(对冲准入停在闸内) →
+        放行 s1 → 轮询 s1 inflight 归零(_attempt finally 结算完,primary 必 done)
+        → 此刻才放行对冲准入。pick 返回时原路已了结,编排必须不落 create_task。
+        """
+        s1 = _src("s1", tpm=1000, est_tokens=400)
+        s2 = _src("s2", tpm=1000, est_tokens=400)
+        inner = InMemoryLimiter(
+            scope="llm",
+            sources={"s1": s1, "s2": s2},
+            global_limits=_NO_GLOBAL,
+            lease_ttl_s=100.0,
+        )
+        limiter = BlockingLimiter(inner, "s2")
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "hedge")]})
+        mw, _, _ = _harness([s1, s2], transport, limiter=limiter)
+        ctx = _ctx()
+        task = asyncio.ensure_future(mw(_req(ctx=ctx)))
+        await transport.entered["s1"].wait()  # 原路在途
+        await limiter.acquiring.wait()  # 对冲准入停在 s2 闸内(触发窗已过)
+        transport.release["s1"].set()  # 原路放行完成
+        while (await inner.source_stats("s1")).inflight != 0:
+            await asyncio.sleep(0.001)  # 结算完 = primary 已 done(同一任务步内返回)
+        limiter.allow.set()  # pick 此刻才返回: primary.done() 已成立
+        resp = await asyncio.wait_for(task, timeout=5)
+        assert resp.content == "ok-s1" and resp.source_name == "s1"
+        assert transport.calls == ["s1"]  # 对冲 HTTP 从未发出(未修前这里会看到 s2)
+        stats = ctx.snapshot()
+        assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False
+        s2_stats = await inner.source_stats("s2")
+        assert s2_stats.inflight == 0 and s2_stats.tpm_used == 0  # 对冲准入按 0 结算释放
+
     async def test_hedge_silent_when_single_source(self):
         """单源 scope: 运行期拿不到异源候选自然静默,行为与不配阈值逐字相同(②)。"""
         transport = HedgeTransport({"s1": [("hang",)]})