"""对冲编排测试(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.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)
_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 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。"""

    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,
    now=None,
):
    """真实 loop 钟装配(对冲计时纪律: 只用 loop 相对时长,不注入 FakeClock)。

    `now` 仅供"注入钟与对冲触发正交"用例注入 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,
        **({"now": now} if now is not None else {}),
    )
    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

    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:
    """异源排除与静默放弃(验收矩阵 ②③)。"""

    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_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",)]})
        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"]  # 恰一轮两路: 计一次预算即耗尽