feat: add opt-in cross-source hedged requests for chat

tests/unit/test_client.py

@@ -144,6 +144,32 @@ class TestChatEndToEnd:
                 await client.chat([{"role": "user", "content": "hi"}], structured="json")
 
 
+class TestHedgeParams:
+    """对冲两个 keyword-only 参数的 client 入口校验(issue #24 H4;计划批次 H)。
+
+    编排行为本身见 tests/unit/test_hedge.py;这里只钉装配面: 入口即校、
+    值域/交叉守卫与 settings 共用同一份 `check_hedge_assembly`。
+    """
+
+    def test_client_hedge_params_entry_validation(self):
+        # 值域: 0/负/非有限当场 ValueError(不经 settings 那道守卫)
+        with pytest.raises(ValueError, match=r"GatewayClient\(hedge_after_s"):
+            _client(hedge_after_s=0)
+        with pytest.raises(ValueError, match="hedge_max_extra"):
+            _client(hedge_max_extra=0)
+        with pytest.raises(ValueError, match="hedge_max_extra"):
+            _client(hedge_max_extra=True)  # bool 不是 int 档位
+        # 交叉: 阈值 ≥ 源 timeout_s(10)= 对冲永不可能触发
+        with pytest.raises(ValueError, match="timeout_s"):
+            _client(hedge_after_s=99.0)
+        # 合法值透传到 RetryMW(单源 scope 的装配 warning 是预期噪音,不断言)
+        client = _client(hedge_after_s=0.05)
+        assert client._hedge_after_s == 0.05
+        assert client._terminal._hedge_after_s == 0.05
+        # 未启用(缺省): 行为逐字等于 1.3.6
+        assert _client()._terminal._hedge_after_s is None
+
+
 class TestGenerationMsClient:
     """裸生成时间的 client 级口径(1.3.7 批次 C2/F)。
 

tests/unit/test_config.py

@@ -1096,3 +1096,165 @@ class TestCallDeadlineConfig:
 
         with pytest.raises(ValueError, match=r"GatewayClient\(call_deadline_s"):
             _client(call_deadline_s=0)
+
+
+class TestHedgeConfig:
+    """`{SCOPE}__HEDGE__AFTER_S`/`{SCOPE}__HEDGE__MAX_EXTRA` 两键与装配守卫(issue #24 H4)。
+
+    对冲默认关闭: 键未设 = None/1,行为逐字等于 1.3.6。守卫四路覆盖
+    (env/直接构造/dataclasses.replace/client 直传),单一定义点是
+    `config.check_hedge_assembly`。
+    """
+
+    def _two_source_env(self, **overrides):
+        """双源 env(同 provider 避免注册表依赖): 隔离单源 warning 的干扰。"""
+        return _env(
+            **{
+                "LLM__QWEN__2__BASE_URL": "https://gw-b.example/v1",
+                "LLM__QWEN__2__API_KEY": "sk-b",
+                "LLM__QWEN__2__MODEL": "qwen-plus",
+                "LLM__QWEN__2__TIMEOUT_S": "90",
+                **overrides,
+            }
+        )
+
+    def test_hedge_keys_from_env_skip_source_loader(self):
+        """两键为 3 段键,天然不被 `_load_sources` 当源字段;`HEDGE` 进保留段防 4 段撞名。"""
+        env = self._two_source_env(**{"LLM__HEDGE__AFTER_S": "8", "LLM__HEDGE__MAX_EXTRA": "2"})
+        with _captured_warnings():  # max_extra>1 的 v1 单路 warning, 本用例不断言它
+            s = GatewaySettings.from_env("LLM", env=env)
+        assert s.hedge_after_s == 8.0
+        assert s.hedge_max_extra == 2
+        assert {src.name for src in s.sources} == {"qwen_1", "qwen_2"}  # HEDGE 键未造源
+        # `HEDGE` 在保留段: `LLM__HEDGE__1__*` 四段键不得造出一个名为 hedge_1 的源
+        env_collision = _env(
+            **{
+                "LLM__HEDGE__1__BASE_URL": "https://gw-c.example/v1",
+                "LLM__HEDGE__1__API_KEY": "sk-c",
+                "LLM__HEDGE__1__MODEL": "m-c",
+                "LLM__HEDGE__1__TIMEOUT_S": "60",
+            }
+        )
+        s2 = GatewaySettings.from_env("LLM", env=env_collision)
+        assert [src.name for src in s2.sources] == ["qwen_1"]
+
+    def test_hedge_keys_unset_mean_disabled(self):
+        """默认关闭: 两键未设 = None/1,且不产生任何 warning。"""
+        with _captured_warnings() as warnings:
+            s = GatewaySettings.from_env("LLM", env=_env())
+        assert s.hedge_after_s is None and s.hedge_max_extra == 1
+        assert not warnings
+
+    def test_hedge_after_s_domain_four_paths(self):
+        """非法值四条装配路全部当场 ValueError(消息须定位得到是哪个键/参数)。"""
+        from tests.unit.test_client import _client
+
+        # 路 1: env(origin 是实际命中的键名)
+        with pytest.raises(ValueError, match="LLM__HEDGE__AFTER_S"):
+            GatewaySettings.from_env("LLM", env=_env(**{"LLM__HEDGE__AFTER_S": "0"}))
+        with pytest.raises(ValueError, match="LLM__HEDGE__AFTER_S"):
+            GatewaySettings.from_env("LLM", env=_env(**{"LLM__HEDGE__AFTER_S": "abc"}))
+        base = GatewaySettings.from_env("LLM", env=_env())
+        # 路 2: 直接构造
+        fields = {f.name: getattr(base, f.name) for f in dataclasses.fields(base)}
+        with pytest.raises(ValueError, match="hedge_after_s"):
+            GatewaySettings(**{**fields, "hedge_after_s": float("nan")})
+        # 路 3: dataclasses.replace
+        with pytest.raises(ValueError, match="hedge_after_s"):
+            dataclasses.replace(base, hedge_after_s=-1)
+        # 路 4: client 直传(不经 settings 那道守卫)
+        with pytest.raises(ValueError, match=r"GatewayClient\(hedge_after_s"):
+            _client(hedge_after_s=0)
+
+    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
+        with pytest.raises(ValueError, match="timeout_s"):
+            GatewaySettings.from_env("LLM", env=env)
+        # 边界内侧合法(89 < 90)
+        s = GatewaySettings.from_env(
+            "LLM", env=self._two_source_env(**{"LLM__HEDGE__AFTER_S": "89"})
+        )
+        assert s.hedge_after_s == 89.0
+
+    def test_hedge_guard_ttft_warns(self):
+        """阈值 ≥ 最小已设 ttft_timeout_s: 流式被看门狗先切,装配期 warning 而非 ValueError。"""
+        env = self._two_source_env(
+            **{
+                "LLM__QWEN__1__TTFT_TIMEOUT_S": "30",
+                "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15",
+                "LLM__HEDGE__AFTER_S": "35",  # ≥ ttft 30, < timeout 90
+            }
+        )
+        with _captured_warnings() as warnings:
+            s = GatewaySettings.from_env("LLM", env=env)
+        assert s.hedge_after_s == 35.0  # warning 不是拒绝: 非流式仍有效
+        assert any("ttft_timeout_s" in m for m in warnings)
+        # 阈值低于看门狗时不告警
+        with _captured_warnings() as warnings2:
+            GatewaySettings.from_env(
+                "LLM",
+                env=self._two_source_env(
+                    **{
+                        "LLM__QWEN__1__TTFT_TIMEOUT_S": "30",
+                        "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15",
+                        "LLM__HEDGE__AFTER_S": "25",
+                    }
+                ),
+            )
+        assert not warnings2
+
+    def test_hedge_guard_single_source_warns(self):
+        """单源 scope 设阈值: 装配期 warning 放行,运行期拿不到候选自然静默。"""
+        with _captured_warnings() as warnings:
+            s = GatewaySettings.from_env("LLM", env=_env(**{"LLM__HEDGE__AFTER_S": "8"}))
+        assert s.hedge_after_s == 8.0
+        assert any("单源" in m for m in warnings)
+
+    def test_hedge_guard_deadline_conflict_raises(self):
+        """阈值 ≥ call_deadline_s: 期限先于对冲触发,对冲形同虚设 → ValueError(§6)。"""
+        env = self._two_source_env(**{"LLM__HEDGE__AFTER_S": "35", "LLM__CALL_DEADLINE_S": "30"})
+        with pytest.raises(ValueError, match="call_deadline_s"):
+            GatewaySettings.from_env("LLM", env=env)
+        # 边界值(恰好相等)同样拒绝
+        with pytest.raises(ValueError, match="call_deadline_s"):
+            GatewaySettings.from_env(
+                "LLM",
+                env=self._two_source_env(
+                    **{"LLM__HEDGE__AFTER_S": "30", "LLM__CALL_DEADLINE_S": "30"}
+                ),
+            )
+        # 阈值 < 期限是合法组合
+        s = GatewaySettings.from_env(
+            "LLM",
+            env=self._two_source_env(**{"LLM__HEDGE__AFTER_S": "29", "LLM__CALL_DEADLINE_S": "30"}),
+        )
+        assert s.hedge_after_s == 29.0 and s.call_deadline_s == 30.0
+
+    def test_hedge_max_extra_v1_cap(self):
+        """max_extra 值域 [1,3] 的四路校验;>1 已接受但 warning 声明 v1 仅单路生效(H5)。"""
+        # 域外值无条件拒绝(即使对冲未启用: 非法值没有"惰性"豁免)
+        with pytest.raises(ValueError, match="hedge_max_extra"):
+            GatewaySettings.from_env("LLM", env=_env(**{"LLM__HEDGE__MAX_EXTRA": "0"}))
+        with pytest.raises(ValueError, match="LLM__HEDGE__MAX_EXTRA"):
+            GatewaySettings.from_env("LLM", env=_env(**{"LLM__HEDGE__MAX_EXTRA": "x"}))
+        with pytest.raises(ValueError, match="hedge_max_extra"):
+            GatewaySettings.from_env("LLM", env=_env(**{"LLM__HEDGE__MAX_EXTRA": "4"}))
+        base = GatewaySettings.from_env("LLM", env=_env())
+        with pytest.raises(ValueError, match="hedge_max_extra"):
+            dataclasses.replace(base, hedge_max_extra=0)
+        # 2/3 接受 + warning: v1 运行期恒单路(对冲任务不再携带首 token 观测,
+        # 行为面由 test_hedge.py 的 ft_events 断言钉住),梯次追加为 H5 预留
+        env = self._two_source_env(**{"LLM__HEDGE__AFTER_S": "8", "LLM__HEDGE__MAX_EXTRA": "2"})
+        with _captured_warnings() as warnings:
+            s = GatewaySettings.from_env("LLM", env=env)
+        assert s.hedge_max_extra == 2
+        assert any("单路" in m for m in warnings)
+
+    def test_from_settings_propagates_hedge_to_client(self):
+        """from_settings 透传: RetryMW 拿到归一化阈值;max_extra 不下传(v1 无消费者)。"""
+        env = self._two_source_env(**{"LLM__HEDGE__AFTER_S": "8"})
+        s = GatewaySettings.from_env("LLM", env=env)
+        client = GatewayClient.from_settings(s)
+        assert client._hedge_after_s == 8.0
+        assert client._terminal._hedge_after_s == 8.0

tests/unit/test_hedge.py

@@ -0,0 +1,456 @@
+"""对冲编排测试(issue #24 设计 §4.6 / 计划批次 G)。
+
+设施纪律(计划 §5): 两源 scope、事件驱动假 transport(每源一对 entered/release
+Event + 可脚本化"先置首 token 再挂起")、**真实 loop 钟**(不注入 FakeClock)、
+`hedge_after_s=0.05`、断言容差 4–10×;取消窗口用 entered 双 Event 栅栏钉死,
+禁 sleep 撞窗口;计时只断言下界与相对比较,不断言精确值。
+"""
+
+import asyncio
+import time
+
+import pytest
+
+from polygateway import CallDeadlineExceeded
+from polygateway.backends.memory.breaker import InMemoryGate
+from polygateway.backends.memory.limiter import InMemoryLimiter
+from polygateway.errors import AllSourcesExhausted, TransientError
+from polygateway.middleware.retry import RetryMW
+from polygateway.sources import SourceCooldownMemo
+from polygateway.types import (
+    BackpressurePolicy,
+    BreakerConfig,
+    ChatRequest,
+    GlobalLimits,
+    RetryPolicy,
+    _CallContext,
+)
+from tests.unit.test_retry import RecordingSelector, StaticSelector, _ok, _src
+
+_BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
+_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
+_HEDGE_AFTER_S = 0.05  # 真实 loop 钟阈值;断言上界取 10×(0.5s)
+
+
+class HedgeTransport:
+    """事件驱动假 transport: 按源剧本精确控制首 token 置位与完成时刻。
+
+    剧本动作(每源一条队列,耗尽后重复最后一项——429 风暴用例需无限供应):
+      ("hang",)                        — 置位该源 entered,挂起直到该源 release(或被取消)
+      ("token_then_hang",)             — 先置 first_token_event 再 hang(首 token 已至)
+      ("succeed", content)             — 立即成功
+      ("succeed_after", delay, content)— 真实 loop 钟睡 delay 后成功
+      ("fail_after", delay, factory)   — 睡 delay 后抛 `factory()` 新造的异常
+    """
+
+    def __init__(self, scripts: dict[str, list[tuple]]):
+        self._scripts = {name: list(actions) for name, actions in scripts.items()}
+        self.calls: list[str] = []
+        # 逐次记录收到的 first_token_event 身份(H5: 对冲路恒为 None,不再梯次)
+        self.ft_events: list[object] = []
+        self.entered = {name: asyncio.Event() for name in scripts}
+        self.release = {name: asyncio.Event() for name in scripts}
+
+    async def complete(
+        self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
+    ):
+        self.calls.append(source.name)
+        self.ft_events.append(first_token_event)
+        actions = self._scripts[source.name]
+        action = actions.pop(0) if len(actions) > 1 else actions[0]
+        kind = action[0]
+        if kind == "hang":
+            self.entered[source.name].set()
+            await self.release[source.name].wait()
+            return _ok(f"ok-{source.name}")
+        if kind == "token_then_hang":
+            if first_token_event is not None:
+                first_token_event.set()
+            self.entered[source.name].set()
+            await self.release[source.name].wait()
+            return _ok(f"ok-{source.name}")
+        if kind == "succeed":
+            return _ok(action[1])
+        if kind == "succeed_after":
+            await asyncio.sleep(action[1])
+            return _ok(action[2])
+        if kind == "fail_after":
+            await asyncio.sleep(action[1])
+            raise action[2]()
+        raise AssertionError(f"未知剧本动作: {action!r}")
+
+
+class RecordingEmitter:
+    """逐次遥测假 emitter: 记录每行的源/错误标签/逻辑调用 ID/attempt call_id。"""
+
+    def __init__(self):
+        self.rows: list[dict] = []
+
+    async def emit_attempt(
+        self,
+        *,
+        request,
+        source,
+        call_id,
+        latency_ms,
+        response,
+        error,
+        reasoning_applies,
+        operation,
+    ):
+        self.rows.append(
+            {
+                "source": source.name,
+                "call_id": call_id,
+                "logical_call_id": request.call_context.logical_call_id
+                if request.call_context is not None
+                else None,
+                "error": error,
+            }
+        )
+
+
+def _harness(
+    sources,
+    transport,
+    *,
+    hedge_after_s=_HEDGE_AFTER_S,
+    max_attempts=3,
+    emitter=None,
+    selector=None,
+    limiter=None,
+    gate=None,
+    stall_window_s=300.0,
+):
+    """真实 loop 钟装配(对冲计时纪律: 只用 loop 相对时长,不注入 FakeClock)。"""
+    limiter = limiter or InMemoryLimiter(
+        scope="llm",
+        sources={s.name: s for s in sources},
+        global_limits=_NO_GLOBAL,
+        lease_ttl_s=100.0,
+    )
+    gate = gate or InMemoryGate(config=_BREAKER)
+    mw = RetryMW(
+        scope="llm",
+        sources=sources,
+        # 固定配置序: 原路恒为 s1、对冲路恒为 s2,断言不依赖选源器内部状态
+        selector=selector if selector is not None else StaticSelector(),
+        limiter=limiter,
+        gate=gate,
+        transport=transport,
+        retry=RetryPolicy(max_attempts=max_attempts, backoff_base_s=0.01, backoff_max_s=0.05),
+        backpressure=BackpressurePolicy(stall_window_s=stall_window_s, poll_interval_s=0.01),
+        quota_full="wait",
+        cooldown_memo=SourceCooldownMemo(),
+        emitter=emitter,
+        hedge_after_s=hedge_after_s,
+    )
+    return mw, limiter, gate
+
+
+def _req(*, stream=False, ctx=None):
+    return ChatRequest(
+        messages=[{"role": "user", "content": "hi"}], stream=stream, call_context=ctx
+    )
+
+
+def _ctx():
+    return _CallContext(now=time.monotonic)
+
+
+class TestHedgeTrigger:
+    """触发两形态(验收矩阵 ①): 非流式纯时间阈值;流式以首 token 未至为判据。"""
+
+    async def test_non_stream_triggers_hedge_and_fast_leg_wins(self):
+        """s1 挂起、s2 即时成功: 对冲截断长尾,赢家为对冲路(⑩: 计时不含触发前等待)。"""
+        s1, s2 = _src("s1"), _src("s2")
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed_after", 0.02, "fast")]})
+        mw, _, _ = _harness([s1, s2], transport)
+        ctx = _ctx()
+        started = time.monotonic()
+        # wait_for 是防挂安全带(红相位无实现时 5s 判负),不是计时断言
+        resp = await asyncio.wait_for(mw(_req(stream=False, ctx=ctx)), timeout=5)
+        elapsed = time.monotonic() - started
+        assert resp.content == "fast" and resp.source_name == "s2"
+        assert elapsed < 10 * _HEDGE_AFTER_S  # 挂起路被对冲截断,而非等到释放
+        stats = ctx.snapshot()
+        assert stats.hedges == 1 and stats.hedge_won is True
+        # 赢家裸生成时间: 下界 = s2 实际 transport 耗时(0.02s 留截断余量),
+        # 且严格小于总时长(不含 0.05s 触发窗等待)
+        assert stats.generation_ms >= 15
+        assert stats.generation_ms < stats.total_latency_ms
+
+    async def test_stream_triggers_only_when_first_token_absent(self):
+        """两例(①): 首 token 未至 → 触发;先置首 token 再挂起 → 不触发。"""
+        # 例一: 流式但首 token 未至,阈值到 → 对冲触发
+        s1, s2 = _src("s1"), _src("s2")
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "fast")]})
+        mw, _, _ = _harness([s1, s2], transport)
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(stream=True, ctx=ctx)), timeout=5)
+        assert resp.source_name == "s2"
+        stats = ctx.snapshot()
+        assert stats.hedges == 1 and stats.hedge_won is True and stats.attempts == 2
+
+        # 例二: 首 token 已至(假 transport 先 set 再挂起)→ 不触发,误杀慢生成即此处
+        transport2 = HedgeTransport({"s1": [("token_then_hang",)], "s2": [("succeed", "other")]})
+        mw2, _, _ = _harness([_src("s1"), _src("s2")], transport2)
+        ctx2 = _ctx()
+
+        async def release_later():
+            await asyncio.sleep(4 * _HEDGE_AFTER_S)  # 4× 余量确认窗口已过
+            transport2.release["s1"].set()
+
+        releaser = asyncio.create_task(release_later())
+        resp2 = await asyncio.wait_for(mw2(_req(stream=True, ctx=ctx2)), timeout=5)
+        await releaser
+        assert resp2.source_name == "s1"
+        assert transport2.calls == ["s1"]  # 对冲从未发出
+        stats2 = ctx2.snapshot()
+        assert stats2.hedges == 0 and stats2.hedge_won is False and stats2.attempts == 1
+
+
+class TestHedgeRouting:
+    """异源排除与静默放弃(验收矩阵 ②③)。"""
+
+    async def test_hedge_goes_to_other_source(self):
+        """对冲请求落在另一源;两 attempt 行共享同一 logical_call_id(②⑤)。"""
+        emitter = RecordingEmitter()
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "hedged")]})
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, emitter=emitter)
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.source_name == "s2"
+        assert transport.calls == ["s1", "s2"]  # 第二请求落在异源
+        # H5 接缝: 原路携带首 token 观测,对冲路恒 None(v1 单路,不再梯次)
+        assert [e is not None for e in transport.ft_events] == [True, False]
+        assert len(emitter.rows) == 2
+        assert {r["logical_call_id"] for r in emitter.rows} == {ctx.logical_call_id}
+        assert emitter.rows[0]["call_id"] != emitter.rows[1]["call_id"]  # 各 attempt 独立 ID
+
+    async def test_hedge_silent_when_no_candidate(self):
+        """异源配额被占满 → 准入失败静默放弃: 不对冲、不抛错、原请求照等(③)。"""
+        s1 = _src("s1")
+        s2 = _src("s2", max_concurrency=1)
+        limiter = InMemoryLimiter(
+            scope="llm", sources={"s1": s1, "s2": s2}, global_limits=_NO_GLOBAL, lease_ttl_s=100.0
+        )
+        held = await limiter.try_acquire("s2", 0)  # 外部预占满 s2 并发
+        assert held is not None
+        try:
+            transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "x")]})
+            mw, _, _ = _harness([s1, s2], transport, limiter=limiter)
+            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
+        finally:
+            await held.release()
+
+    async def test_hedge_silent_when_single_source(self):
+        """单源 scope: 运行期拿不到异源候选自然静默,行为与不配阈值逐字相同(②)。"""
+        transport = HedgeTransport({"s1": [("hang",)]})
+        mw, _, _ = _harness([_src("s1")], transport)
+        ctx = _ctx()
+        task = asyncio.ensure_future(mw(_req(ctx=ctx)))
+        await transport.entered["s1"].wait()
+        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.content == "ok-s1"
+        stats = ctx.snapshot()
+        assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False
+
+
+class TestHedgeSettlementAndSignals:
+    """赢输记账与熔断/健康信号(验收矩阵 ④⑤;设计 §3 关键判断: 挂起 ≠ 源死亡)。"""
+
+    async def test_winner_settles_actual_loser_keeps_est(self):
+        """赢家按真实 usage 结算;输家取消落 1.3.6 S3 格: est 预扣保留(④)。"""
+        s1 = _src("s1", tpm=1000, est_tokens=400)
+        s2 = _src("s2", tpm=1000, est_tokens=400)
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
+        mw, limiter, _ = _harness([s1, s2], transport)
+        resp = await asyncio.wait_for(mw(_req()), timeout=5)
+        assert resp.source_name == "s2"
+        winner_stats = await limiter.source_stats("s2")
+        loser_stats = await limiter.source_stats("s1")
+        assert winner_stats.tpm_used == 15  # 预扣 400,实测 10+5 → settle 后只记 15
+        assert loser_stats.tpm_used == 400  # 输家 est 保留(可能被上游计费,保守下限)
+        assert winner_stats.inflight == 0 and loser_stats.inflight == 0
+
+    async def test_loser_row_labelled_hedge_cancelled(self):
+        """输家 attempt 行 error=='hedge_cancelled',赢家行无 error,同行逻辑调用(④⑤)。
+
+        终态行是 client 级语义且成功调用本就不写终态行(emit_terminal_once 只在
+        异常/取消路径触发),MW 级可观测面即这两条 attempt 行。
+        """
+        emitter = RecordingEmitter()
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, emitter=emitter)
+        ctx = _ctx()
+        await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert len(emitter.rows) == 2
+        loser = next(r for r in emitter.rows if r["source"] == "s1")
+        winner = next(r for r in emitter.rows if r["source"] == "s2")
+        assert loser["error"] == "hedge_cancelled"
+        assert winner["error"] is None
+        assert loser["logical_call_id"] == winner["logical_call_id"] == ctx.logical_call_id
+
+    async def test_loser_does_not_feed_breaker(self):
+        """输家取消不喂熔断失败计数、不喂健康分;赢家照常 record_success(④)。"""
+        selector = RecordingSelector()
+        gate = InMemoryGate(config=_BREAKER)
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, selector=selector, gate=gate)
+        await asyncio.wait_for(mw(_req()), timeout=5)
+        gate_s1 = gate._gates["s1"]
+        assert gate_s1.a0 + gate_s1.a1 == 0  # 熔断失败率窗口无样本
+        assert selector.outcomes == [("s2", True)]  # 健康喂数只有赢家的成功
+        assert (await gate.try_enter("s1", "w")).allowed  # 挂起源未被标记
+
+    async def test_attempts_two_and_no_task_leak(self):
+        """快照 attempts==2(含输家);返回后无本调用残留任务(⑤)。"""
+        before = asyncio.all_tasks()
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport)
+        ctx = _ctx()
+        await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert ctx.snapshot().attempts == 2
+        assert asyncio.all_tasks() == before
+
+
+class TestHedgeCancellation:
+    """取消穿透(⑦)与期限组合(⑧): 两任务同消、不 shield、不留后台任务。"""
+
+    async def test_external_cancel_cancels_both_legs(self):
+        """两路均在途时外部取消: CancelledError 上抛,两 permit 释放。
+
+        输家标记只在赢家产生后才置位——外部取消下没有赢家,两行都是普通
+        "cancelled"(⑦;竞速误贴属设计 §4.5 已批准残留)。
+        """
+        emitter = RecordingEmitter()
+        s1, s2 = _src("s1"), _src("s2")
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("hang",)]})
+        mw, limiter, _ = _harness([s1, s2], transport, emitter=emitter)
+        task = asyncio.ensure_future(mw(_req()))
+        # 双 Event 栅栏: 确认对冲已触发、两路均在途,再取消(禁 sleep 猜窗口)
+        await transport.entered["s1"].wait()
+        await transport.entered["s2"].wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert {r["source"]: r["error"] for r in emitter.rows} == {
+            "s1": "cancelled",
+            "s2": "cancelled",
+        }
+        assert (await limiter.source_stats("s1")).inflight == 0
+        assert (await limiter.source_stats("s2")).inflight == 0
+
+    async def test_deadline_cuts_hedged_tree(self):
+        """client 级 call_deadline_s=0.2 + 两路挂起 → CallDeadlineExceeded,permit 全释放(⑧)。"""
+        from tests.unit.test_client import _client
+
+        s1, s2 = _src("s1"), _src("s2")
+        limiter = InMemoryLimiter(
+            scope="llm", sources={"s1": s1, "s2": s2}, global_limits=_NO_GLOBAL
+        )
+        transport = HedgeTransport({"s1": [("hang",)], "s2": [("hang",)]})
+        client = _client(
+            sources=[s1, s2],
+            transport=transport,
+            limiter=limiter,
+            call_deadline_s=0.2,
+            hedge_after_s=_HEDGE_AFTER_S,
+        )
+        async with client:
+            with pytest.raises(CallDeadlineExceeded):
+                await client.chat([{"role": "user", "content": "hi"}], stream=False)
+        assert transport.calls == ["s1", "s2"]  # 期限截止前对冲确已触发
+        assert (await limiter.source_stats("s1")).inflight == 0
+        assert (await limiter.source_stats("s2")).inflight == 0
+
+
+class TestHedgeWinnerAdjudication:
+    """赢家裁定(⑩): 取快者,含原路后发先至的对称面。"""
+
+    async def test_primary_late_success_wins_back(self):
+        """s1 挂 0.3s(6× 阈值)后成功、s2 对冲路在途: 原路先完成 → 原路赢。"""
+        emitter = RecordingEmitter()
+        transport = HedgeTransport({"s1": [("succeed_after", 0.3, "late")], "s2": [("hang",)]})
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, emitter=emitter)
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.content == "late" and resp.source_name == "s1"
+        stats = ctx.snapshot()
+        assert stats.hedges == 1 and stats.hedge_won is False
+        # 裸生成时间为原路那次 transport 时长(≈300ms,只断言下界与相对关系)
+        assert 250 <= stats.generation_ms <= stats.total_latency_ms
+        loser = next(r for r in emitter.rows if r["source"] == "s2")
+        assert loser["error"] == "hedge_cancelled"  # 在途对冲路被裁为输家
+
+
+class TestHedgeFailureCombination:
+    """两败汇合(H6): 只计一次重试预算;429 分账逐字沿用 attempt 级机制。"""
+
+    async def test_both_fail_counts_budget_once(self):
+        """两路 Transient: max_attempts=2 时恰进第二轮(两败只计一次),第二轮两败后才耗尽。"""
+        transport = HedgeTransport(
+            {
+                "s1": [("fail_after", 0.1, lambda: TransientError("p", source_name="s1"))],
+                "s2": [("fail_after", 0.12, lambda: TransientError("h", source_name="s2"))],
+            }
+        )
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, max_attempts=2)
+        ctx = _ctx()
+        with pytest.raises(AllSourcesExhausted) as ei:
+            await mw(_req(ctx=ctx))
+        assert ei.value.reason == "retry_exhausted"
+        # 若两败计两次预算,第一轮即耗尽,这些调用根本不会发生
+        assert transport.calls == ["s1", "s2", "s1", "s2"]
+        # 两败轮次同样登记对冲路数(设计 §4.5: 实际并发发出即计)
+        assert ctx.snapshot().hedges == 2
+
+    async def test_both_429_refund_no_budget(self):
+        """两路皆 429: 免预算且耗时退 stall 账——小 stall 窗下终局 stalled 而非耗尽。"""
+
+        def _429():
+            return TransientError("throttled", status_code=429, retry_after_s=0.01)
+
+        transport = HedgeTransport(
+            {"s1": [("fail_after", 0.1, _429)], "s2": [("fail_after", 0.12, _429)]}
+        )
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, max_attempts=1, stall_window_s=0.3)
+        with pytest.raises(AllSourcesExhausted) as ei:
+            await mw(_req())
+        # max_attempts=1: 任一路计预算都会当场 retry_exhausted;
+        # 两 429 免预算 → 循环到 stall 窗口判死
+        assert ei.value.reason == "stalled"
+
+    async def test_mixed_429_and_failure_counts_budget(self):
+        """一路 429 一路 Transient → 计一次预算、不退还 stall 账(_combine_failures)。"""
+        transport = HedgeTransport(
+            {
+                "s1": [
+                    (
+                        "fail_after",
+                        0.1,
+                        lambda: TransientError("rl", status_code=429, retry_after_s=0.01),
+                    )
+                ],
+                "s2": [("fail_after", 0.12, lambda: TransientError("boom", source_name="s2"))],
+            }
+        )
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, max_attempts=1, stall_window_s=0.3)
+        with pytest.raises(AllSourcesExhausted) as ei:
+            await mw(_req())
+        assert ei.value.reason == "retry_exhausted"
+        assert transport.calls == ["s1", "s2"]  # 恰一轮两路: 计一次预算即耗尽