"""背压 stall 双条件判死 + poll jitter + 记账侧降级(M2 设计 §4/§10)。

保真蓝本 CHS governance.py:200-285: local_waited(本地 monotonic,调用级
累计不重置)与 progress_age(全局活性)**同时**超 stall_window 才判死;
poll 间隔带 [0.5p, 1.0p] jitter 防惊群。
"""

import asyncio

import pytest

from polygateway.backends.memory.breaker import InMemoryGate
from polygateway.backends.memory.limiter import InMemoryLimiter
from polygateway.errors import (
    AllSourcesExhausted,
    GatewayUnavailableError,
    GovernanceBackendError,
    SourceNotConfiguredError,
    TransientError,
)
from polygateway.middleware.retry import RetryMW, backoff_delay
from polygateway.sources import RoundRobinSelector, SourceCooldownMemo
from polygateway.types import (
    BackpressurePolicy,
    BreakerConfig,
    ChatRequest,
    GlobalLimits,
    RetryPolicy,
)
from tests.contracts.conftest import FakeClock, make_source
from tests.unit.test_retry import FakeTransport, _ok

_BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
_REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}])
_STALL = 300.0


class BoundedSleep:
    """每次 sleep 执行注入的副作用;超过上限仍在轮询 = 判死逻辑失效,炸出而非死循环。"""

    def __init__(self, side_effect=None, limit: int = 10):
        self.delays: list[float] = []
        self._side_effect = side_effect
        self._limit = limit

    async def __call__(self, seconds: float) -> None:
        self.delays.append(seconds)
        if len(self.delays) > self._limit:
            raise RuntimeError(f"超过 {self._limit} 次轮询仍未判死/未获 permit")
        if self._side_effect is not None:
            await self._side_effect(len(self.delays))


def _mw(sources, limiter, script, *, clock, sleep, rng=lambda: 0.0, quota_full="wait", gate=None):
    return RetryMW(
        scope="llm",
        sources=sources,
        selector=RoundRobinSelector(),
        limiter=limiter,
        gate=gate or InMemoryGate(config=_BREAKER, now=clock),
        transport=FakeTransport(script),
        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,
        cooldown_memo=SourceCooldownMemo(now=clock),
        emitter=None,
        now=clock,
        sleep=sleep,
        rng=rng,
    )


def _blocked_limiter(clock):
    """单并发源被外部占满 → RetryMW 进入 wait 轮询分支。"""
    src = make_source(max_concurrency=1)
    limiter = InMemoryLimiter(
        scope="llm",
        sources={"s1": src},
        global_limits=_NO_GLOBAL,
        lease_ttl_s=10_000.0,
        now=clock,
    )
    return src, limiter


class TestStallQuadrants:
    async def test_both_windows_exceeded_raises_stalled(self):
        """双超窗: 本地等待与全局无进展同时 > stall_window → stalled。"""
        clock = FakeClock()
        src, limiter = _blocked_limiter(clock)
        _held = await limiter.try_acquire("s1", 0)

        async def advance(_n):
            clock.advance(_STALL + 100)

        mw = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(advance))
        with pytest.raises(AllSourcesExhausted) as ei:
            await mw(_REQ)
        assert ei.value.reason == "stalled"
        assert ei.value.per_source_reasons == {"s1": "rate_limited"}

    async def test_local_exceeded_but_global_fresh_keeps_waiting(self):
        """仅本地超窗: 别人一直在出餐 → 不判死,等到 permit 后正常成功。"""
        clock = FakeClock()
        src, limiter = _blocked_limiter(clock)
        held = await limiter.try_acquire("s1", 0)

        async def advance_and_feed(n):
            clock.advance(_STALL + 100)  # 本地远超窗
            await limiter.mark_progress()  # 但全局刚出过餐
            if n >= 3:
                await held.release()  # 第 3 轮让出 permit

        mw = _mw([src], limiter, [_ok()], clock=clock, sleep=BoundedSleep(advance_and_feed))
        resp = await mw(_REQ)
        assert resp.content == "ok"

    async def test_global_stale_but_local_fresh_keeps_waiting(self):
        """仅全局超窗(从未出餐 age=inf): 本地才刚开始等 → 不判死。"""
        clock = FakeClock()
        src, limiter = _blocked_limiter(clock)
        held = await limiter.try_acquire("s1", 0)

        async def release_soon(n):
            if n >= 2:  # 本地累计 poll 极短,未超窗
                await held.release()

        mw = _mw([src], limiter, [_ok()], clock=clock, sleep=BoundedSleep(release_soon))
        resp = await mw(_REQ)
        assert resp.content == "ok"

    async def test_poll_jitter_bounds(self):
        """wait 轮询间隔 ∈ [0.5p, 1.0p](CHS governance.py:283-285)。"""
        clock = FakeClock()
        src, limiter = _blocked_limiter(clock)
        held = await limiter.try_acquire("s1", 0)

        async def release_soon(n):
            if n >= 2:
                await held.release()

        for rng_v, expected in ((0.0, 0.005), (1.0, 0.01)):
            clock2 = FakeClock()
            src2, limiter2 = _blocked_limiter(clock2)
            held2 = await limiter2.try_acquire("s1", 0)

            async def release2(n, _h=held2):
                if n >= 2:
                    await _h.release()

            sleep = BoundedSleep(release2)
            mw = _mw([src2], limiter2, [_ok()], clock=clock2, sleep=sleep, rng=lambda v=rng_v: v)
            await mw(_REQ)
            assert sleep.delays[0] == pytest.approx(expected)

    async def test_fail_fast_unaffected(self):
        """fail_fast 路径回归: 不进入 stall 判定,立即 quota_exhausted。"""
        clock = FakeClock()
        src, limiter = _blocked_limiter(clock)
        _held = await limiter.try_acquire("s1", 0)
        mw = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), quota_full="fail_fast")
        with pytest.raises(AllSourcesExhausted) as ei:
            await mw(_REQ)
        assert ei.value.reason == "quota_exhausted"

    async def test_cancellation_pierces_wait_loop(self):
        """stall 等待中的取消穿透(铁律): sleep 可取消,任务立即终止。"""
        clock = FakeClock()
        src, limiter = _blocked_limiter(clock)
        _held = await limiter.try_acquire("s1", 0)
        mw = _mw([src], limiter, [], clock=clock, sleep=asyncio.sleep)
        task = asyncio.create_task(mw(_REQ))
        await asyncio.sleep(0.03)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task


class _GateSuccessBroken(InMemoryGate):
    async def record_success(self, entry):
        raise GovernanceBackendError("redis 抖动", scope="llm")


class _GateFailureBroken(InMemoryGate):
    async def record_failure(self, entry, reason, force_open):
        raise GovernanceBackendError("redis 抖动", scope="llm")


class _LimiterProgressBroken(InMemoryLimiter):
    async def mark_progress(self):
        raise GovernanceBackendError("redis 抖动", scope="llm")


class TestAccountingDegradation:
    """记账侧降级(设计 §10,ARCH §7.3 勘误): 调用已完成,写回失败不冒泡。"""

    async def test_record_success_failure_does_not_lose_response(self):
        clock = FakeClock()
        src = make_source()
        limiter = InMemoryLimiter(
            scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock
        )
        gate = _GateSuccessBroken(config=_BREAKER, now=clock)
        mw = _mw([src], limiter, [_ok()], clock=clock, sleep=BoundedSleep(), gate=gate)
        resp = await mw(_REQ)
        assert resp.content == "ok"  # 真实成功响应不因记账失败被丢弃

    async def test_mark_progress_failure_does_not_lose_response(self):
        clock = FakeClock()
        src = make_source()
        limiter = _LimiterProgressBroken(
            scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock
        )
        mw = _mw([src], limiter, [_ok()], clock=clock, sleep=BoundedSleep())
        resp = await mw(_REQ)
        assert resp.content == "ok"

    async def test_record_failure_failure_does_not_mask_retry(self):
        """失败记账挂掉 → 原始尝试异常不被掩盖,重试照常换发并成功。"""
        clock = FakeClock()
        src = make_source()
        limiter = InMemoryLimiter(
            scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock
        )
        gate = _GateFailureBroken(config=_BREAKER, now=clock)
        script = [TransientError("boom", status_code=500), _ok()]
        mw = _mw([src], limiter, script, clock=clock, sleep=BoundedSleep(), gate=gate)
        resp = await mw(_REQ)
        assert resp.content == "ok"


class TestBackoffExtraction:
    """T5 提取的模块级纯函数(T9 Embedding 复用)。"""

    def test_formula_matches_policy(self):
        policy = RetryPolicy(max_attempts=3, backoff_base_s=2.0, backoff_max_s=30.0)
        exc = TransientError("x", status_code=500)
        assert backoff_delay(policy, 1, exc, lambda: 0.0) == pytest.approx(1.0)  # 2*0.5
        assert backoff_delay(policy, 1, exc, lambda: 1.0) == pytest.approx(3.0)  # 2*1.5
        assert backoff_delay(policy, 10, exc, lambda: 1.0) == pytest.approx(45.0)  # 封顶 30*1.5

    def test_retry_after_hint_wins_when_larger(self):
        policy = RetryPolicy(max_attempts=3, backoff_base_s=2.0, backoff_max_s=30.0)
        exc = TransientError("x", status_code=429, retry_after_s=7.5)
        assert backoff_delay(policy, 1, exc, lambda: 0.0) == pytest.approx(7.5)


class TestQuotaGateProgressAge:
    async def test_passthrough_and_wrap(self):
        from polygateway.middleware.ratelimit import QuotaGate

        class _L:
            async def progress_age_s(self):
                return 12.5

        class _Broken:
            async def progress_age_s(self):
                raise OSError("down")

        assert await QuotaGate(_L(), scope="llm").progress_age_s() == 12.5
        with pytest.raises(GovernanceBackendError):
            await QuotaGate(_Broken(), scope="llm").progress_age_s()


class TestUnknownSourceIsAssemblyDefect:
    """未知源 = 限流后端的源名单与治理循环对不上,是装配缺陷不是后端故障。

    两个后端行为必须一致(Redis 版对应用例在 `test_redis_key_layout.py::
    TestConversions::test_unknown_source_rejected`);内存版此前无覆盖,
    该分支从未被测过(issue #7 §3.4)。
    """

    def test_memory_limiter_rejects_unknown_source(self):
        limiter = InMemoryLimiter(
            scope="llm", sources={"s1": make_source("s1")}, global_limits=_NO_GLOBAL
        )
        with pytest.raises(SourceNotConfiguredError) as ei:
            limiter._cfg("nope")
        # 关键: 若归入 scope 级家族,配置写错的任务会永远延期重投、永不进死信
        assert not isinstance(ei.value, GatewayUnavailableError)


class TestGateFailuresReachCallersAsScopeLevel:
    """三条闸门泄漏路径必须以 scope 级不可用的形态到达调用方(issue #7)。

    记账路径由 `_record_quietly` 降级为 warning,但闸门路径没有那层包裹,会一路
    抛给调用方。只写 `except GatewayUnavailableError` 的调用方此前接不住,后果
    是 Redis 抖一下就让积压任务烧掉业务失败预算进死信——而那是运维重启即可恢复
    的故障。三条路径逐一钉住,防止将来任何一条被漏掉。
    """

    async def test_try_acquire_failure_is_scope_level(self):
        from polygateway.middleware.ratelimit import QuotaGate

        class _Broken:
            async def try_acquire(self, name, est):
                raise OSError("down")

        with pytest.raises(GatewayUnavailableError) as ei:
            await QuotaGate(_Broken(), scope="LLM").try_acquire(make_source("s1"))
        assert ei.value.scope == "llm"
        assert ei.value.reason == "governance_backend_down"
        assert ei.value.retry_after_s > 0  # 0 会让积压任务零延迟冲击已挂的后端

    async def test_try_enter_failure_is_scope_level(self):
        from polygateway.middleware.breaker import BreakerGate

        class _Broken:
            async def try_enter(self, name, owner):
                raise OSError("down")

        with pytest.raises(GatewayUnavailableError) as ei:
            await BreakerGate(_Broken(), scope="LLM").try_enter(make_source("s1"), "owner")
        assert ei.value.scope == "llm"
        assert ei.value.reason == "governance_backend_down"

    async def test_progress_age_failure_is_scope_level(self):
        from polygateway.middleware.ratelimit import QuotaGate

        class _Broken:
            async def progress_age_s(self):
                raise OSError("down")

        with pytest.raises(GatewayUnavailableError) as ei:
            await QuotaGate(_Broken(), scope="LLM").progress_age_s()
        assert ei.value.scope == "llm"
        assert ei.value.reason == "governance_backend_down"