From 4a176b6220af376233977718f35d0b30792280bb Mon Sep 17 00:00:00 2001
From: iomgaa <iomgaaycz@gmail.com>
Date: Mon, 20 Jul 2026 06:54:07 -0400
Subject: [PATCH] feat: add in-memory limiter and breaker satisfying backend
 contracts

---
 src/polygateway/backends/memory/breaker.py | 160 +++++++++++++++++++
 src/polygateway/backends/memory/limiter.py | 170 +++++++++++++++++++++
 tests/__init__.py                          |   0
 tests/contracts/conftest.py                |  61 ++++++++
 tests/contracts/test_breaker_contract.py   | 151 ++++++++++++++++++
 tests/contracts/test_limiter_contract.py   | 103 +++++++++++++
 tests/e2e/__init__.py                      |   0
 tests/integration/__init__.py              |   0
 tests/unit/__init__.py                     |   0
 9 files changed, 645 insertions(+)
 create mode 100644 src/polygateway/backends/memory/breaker.py
 create mode 100644 src/polygateway/backends/memory/limiter.py
 create mode 100644 tests/__init__.py
 create mode 100644 tests/contracts/conftest.py
 create mode 100644 tests/contracts/test_breaker_contract.py
 create mode 100644 tests/contracts/test_limiter_contract.py
 create mode 100644 tests/e2e/__init__.py
 create mode 100644 tests/integration/__init__.py
 create mode 100644 tests/unit/__init__.py

diff --git a/src/polygateway/backends/memory/breaker.py b/src/polygateway/backends/memory/breaker.py
new file mode 100644
index 0000000..129669c
--- /dev/null
+++ b/src/polygateway/backends/memory/breaker.py
@@ -0,0 +1,160 @@
+"""进程内熔断门: CHS gate 契约的内存实现(D3 双后端)。
+
+状态机蓝本 VT `adapters/breaker.py`(闭路→阈值开路→冷却半开→单探针),
+契约形态承 CHS `provider_gate.py`: 半开探针是**带 TTL 的租约**(持有者
+死亡后可被接管,防"探针永远在路上"死锁),写回经 epoch fencing 拒绝
+旧世代污染。epoch 在每次进入 OPEN 时递增。时钟构造注入,纯确定性可测。
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+from polygateway.ports import GateDecision, GateState, GateUpdate
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from polygateway.types import BreakerConfig
+
+
+@dataclass
+class _SourceGate:
+    """单源的门控可变状态。"""
+
+    state: GateState = GateState.CLOSED
+    fails: int = 0
+    epoch: int = 0
+    open_until: float = 0.0
+    probe_owner: str | None = None
+    probe_expires: float = 0.0
+    reasons: dict[str, str] = field(default_factory=dict)
+
+
+class InMemoryGate:
+    """按 source_name 分别计数的进程内熔断门。"""
+
+    def __init__(self, *, config: BreakerConfig, now: Callable[[], float] = time.monotonic) -> None:
+        self._cfg = config
+        self._now = now
+        self._gates: dict[str, _SourceGate] = {}
+
+    def _gate(self, source_name: str) -> _SourceGate:
+        if not source_name.strip():
+            raise ValueError("source_name 不能为空")
+        return self._gates.setdefault(source_name, _SourceGate())
+
+    def _grant_probe(self, g: _SourceGate, source_name: str, owner: str) -> GateDecision:
+        g.state = GateState.HALF_OPEN
+        g.probe_owner = owner
+        g.probe_expires = self._now() + self._cfg.probe_ttl_s
+        return GateDecision(
+            source_name=source_name, allowed=True, state=GateState.HALF_OPEN,
+            epoch=g.epoch, is_probe=True, probe_owner=owner, retry_after_s=0.0,
+        )
+
+    async def try_enter(self, source_name: str, owner: str) -> GateDecision:
+        """健康普通准入;冷却到期/探针租约过期时原子授予唯一探针。"""
+        if not owner.strip():
+            raise ValueError("owner 不能为空")
+        g = self._gate(source_name)
+        now = self._now()
+        if g.state is GateState.CLOSED:
+            return GateDecision(
+                source_name=source_name, allowed=True, state=GateState.CLOSED,
+                epoch=g.epoch, is_probe=False, probe_owner=None, retry_after_s=0.0,
+            )
+        if g.state is GateState.OPEN:
+            if now >= g.open_until:
+                return self._grant_probe(g, source_name, owner)
+            return GateDecision(
+                source_name=source_name, allowed=False, state=GateState.OPEN,
+                epoch=g.epoch, is_probe=False, probe_owner=None,
+                retry_after_s=g.open_until - now,
+            )
+        # HALF_OPEN: 探针在途;租约过期则接管,否则拒绝(防惊群)
+        if now >= g.probe_expires:
+            return self._grant_probe(g, source_name, owner)
+        return GateDecision(
+            source_name=source_name, allowed=False, state=GateState.HALF_OPEN,
+            epoch=g.epoch, is_probe=False, probe_owner=None,
+            retry_after_s=g.probe_expires - now,
+        )
+
+    def _fenced(self, g: _SourceGate, entry: GateDecision) -> bool:
+        """写回资格: 世代一致;探针写回还要求 owner 仍在位(CHS fencing 同款)。"""
+        if not entry.allowed:
+            raise ValueError("被拒决定不得写回")
+        if entry.epoch != g.epoch:
+            return False
+        return not (
+            entry.is_probe
+            and (g.state is not GateState.HALF_OPEN or g.probe_owner != entry.probe_owner)
+        )
+
+    def _snapshot(self, g: _SourceGate, applied: bool) -> GateUpdate:
+        return GateUpdate(
+            applied=applied, state=g.state, epoch=g.epoch, failure_count=g.fails,
+            retry_after_s=max(0.0, g.open_until - self._now()) if g.state is GateState.OPEN else 0.0,
+        )
+
+    def _open(self, g: _SourceGate, reason: str) -> None:
+        g.state = GateState.OPEN
+        g.epoch += 1  # 世代推进: 旧 entry 的迟到写回自此被 fencing 拒绝
+        g.open_until = self._now() + self._cfg.cooldown_s
+        g.fails = max(g.fails, self._cfg.fail_threshold)
+        g.probe_owner = None
+        g.probe_expires = 0.0
+
+    async def record_success(self, entry: GateDecision) -> GateUpdate:
+        g = self._gate(entry.source_name)
+        if not self._fenced(g, entry):
+            return self._snapshot(g, applied=False)
+        g.state = GateState.CLOSED
+        g.fails = 0
+        g.probe_owner = None
+        g.probe_expires = 0.0
+        return self._snapshot(g, applied=True)
+
+    async def record_failure(self, entry: GateDecision, reason: str, force_open: bool) -> GateUpdate:
+        g = self._gate(entry.source_name)
+        if not self._fenced(g, entry):
+            return self._snapshot(g, applied=False)
+        if entry.is_probe or force_open:
+            self._open(g, reason)  # 探针失败重开 / SourceDead 一击即熔
+            return self._snapshot(g, applied=True)
+        g.fails += 1
+        if g.fails >= self._cfg.fail_threshold:
+            self._open(g, reason)
+        return self._snapshot(g, applied=True)
+
+    async def release_probe(self, entry: GateDecision) -> GateUpdate:
+        """探针无果归还(如取消): 源保持可接管状态让下一 caller 接手;幂等。"""
+        if not (entry.allowed and entry.is_probe and entry.probe_owner is not None):
+            raise ValueError("release_probe 只接受在途探针决定")
+        g = self._gate(entry.source_name)
+        if not self._fenced(g, entry):
+            return self._snapshot(g, applied=False)
+        g.state = GateState.OPEN
+        g.open_until = self._now()  # 立即可被下一 caller 以探针身份接管
+        g.probe_owner = None
+        g.probe_expires = 0.0
+        return self._snapshot(g, applied=True)
+
+    async def retry_after_s(self, sources: tuple[str, ...]) -> float:
+        """集合中最早可尝试时间;健康/到期返回 0。"""
+        if not sources:
+            raise ValueError("sources 不能为空")
+        now = self._now()
+        waits = []
+        for name in sources:
+            g = self._gate(name)
+            if g.state is GateState.OPEN:
+                waits.append(max(0.0, g.open_until - now))
+            elif g.state is GateState.HALF_OPEN:
+                waits.append(max(0.0, g.probe_expires - now))
+            else:
+                waits.append(0.0)
+        return min(waits)
diff --git a/src/polygateway/backends/memory/limiter.py b/src/polygateway/backends/memory/limiter.py
new file mode 100644
index 0000000..6e124ce
--- /dev/null
+++ b/src/polygateway/backends/memory/limiter.py
@@ -0,0 +1,170 @@
+"""进程内限流后端: 与 Redis 版同一契约的六道闸实现(D3 双后端)。
+
+语义蓝本 CHS `app/coordination/limiter.py`: 并发 = 带 TTL 的租约(持有者
+死亡后过期回收);RPM/TPM = 分钟滑动窗口计数;TPM 入场按 est 预扣,settle
+按实际结算多退少补且退款落 acquire 时的窗口。检查-占用在单次同步段内完成
+(无 await 穿插),单事件循环下天然原子;本实现不跨进程,是单进程部署的
+正确答案(多 worker 用 M2 Redis 后端)。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+import uuid
+from typing import TYPE_CHECKING
+
+from polygateway.errors import GovernanceBackendError
+from polygateway.types import GlobalLimits, SourceConfig, SourceStats
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+_GLOBAL = "__global__"
+_WINDOW_S = 60.0
+
+
+class _MemoryPermit:
+    """入场许可;release/settle 幂等(CHS _RedisPermit 同款 flag 语义)。"""
+
+    def __init__(self, limiter: InMemoryLimiter, source: str, lease_id: str, est: int, window: int) -> None:
+        self._limiter = limiter
+        self._source = source
+        self._lease_id = lease_id
+        self._est = est
+        self._window = window
+        self._released = False
+        self._settled = False
+
+    async def release(self) -> None:
+        if self._released:
+            return
+        self._released = True
+        self._limiter._release_lease(self._source, self._lease_id)
+
+    async def settle(self, actual_tokens: int) -> None:
+        if self._settled:
+            return
+        self._settled = True
+        delta = actual_tokens - self._est
+        if delta:
+            self._limiter._settle_tpm(self._source, delta, self._window)
+
+
+class InMemoryLimiter:
+    """六道闸: 全局/单源 × 并发/RPM/TPM;限额 0 = 该闸不启用。"""
+
+    def __init__(
+        self,
+        *,
+        scope: str,
+        sources: dict[str, SourceConfig],
+        global_limits: GlobalLimits,
+        lease_ttl_s: float = 1500.0,
+        now: Callable[[], float] = time.monotonic,
+        sleep: Callable[[float], object] = asyncio.sleep,
+        poll_interval_s: float = 0.05,
+    ) -> None:
+        if lease_ttl_s <= 0 or poll_interval_s <= 0:
+            raise ValueError("lease_ttl_s 与 poll_interval_s 必须 > 0")
+        self._scope = scope
+        self._sources = dict(sources)
+        self._global = global_limits
+        self._lease_ttl_s = lease_ttl_s
+        self._now = now
+        self._sleep = sleep
+        self._poll_interval_s = poll_interval_s
+        # source → lease_id → 过期时刻
+        self._leases: dict[str, dict[str, float]] = {name: {} for name in self._sources}
+        # (source|__global__, window_id) → 计数
+        self._rpm: dict[tuple[str, int], int] = {}
+        self._tpm: dict[tuple[str, int], int] = {}
+        self._progress_at: float | None = None
+
+    # —— 内部原语(同步,单事件循环下原子)——
+
+    def _cfg(self, source_key: str) -> SourceConfig:
+        cfg = self._sources.get(source_key)
+        if cfg is None:
+            raise GovernanceBackendError(f"未知源 {source_key!r}(scope={self._scope})")
+        return cfg
+
+    def _window(self) -> int:
+        return int(self._now() / _WINDOW_S)
+
+    def _purge_leases(self) -> None:
+        now = self._now()
+        for leases in self._leases.values():
+            expired = [lid for lid, expiry in leases.items() if expiry <= now]
+            for lid in expired:
+                del leases[lid]
+
+    def _inflight(self, source_key: str) -> int:
+        return len(self._leases[source_key])
+
+    def _inflight_total(self) -> int:
+        return sum(len(leases) for leases in self._leases.values())
+
+    def _release_lease(self, source_key: str, lease_id: str) -> None:
+        self._leases[source_key].pop(lease_id, None)
+
+    def _settle_tpm(self, source_key: str, delta: int, window: int) -> None:
+        # 退款/补记落 acquire 时的窗口,计数下限 0(CHS Lua 同款)
+        for key in (source_key, _GLOBAL):
+            slot = (key, window)
+            self._tpm[slot] = max(0, self._tpm.get(slot, 0) + delta)
+
+    def _gates_pass(self, cfg: SourceConfig, est_tokens: int, window: int) -> bool:
+        checks = (
+            (self._global.max_concurrency, self._inflight_total() + 1),
+            (cfg.max_concurrency, self._inflight(cfg.name) + 1),
+            (self._global.rpm, self._rpm.get((_GLOBAL, window), 0) + 1),
+            (cfg.rpm, self._rpm.get((cfg.name, window), 0) + 1),
+            (self._global.tpm, self._tpm.get((_GLOBAL, window), 0) + est_tokens),
+            (cfg.tpm, self._tpm.get((cfg.name, window), 0) + est_tokens),
+        )
+        return all(limit <= 0 or would_be <= limit for limit, would_be in checks)
+
+    # —— 契约方法 ——
+
+    async def try_acquire(self, source_key: str, est_tokens: int) -> _MemoryPermit | None:
+        """非阻塞准入: 六闸全过才占用;任一满则返回 None 且零副作用。"""
+        cfg = self._cfg(source_key)
+        self._purge_leases()
+        window = self._window()
+        if not self._gates_pass(cfg, est_tokens, window):
+            return None
+        lease_id = uuid.uuid4().hex
+        self._leases[source_key][lease_id] = self._now() + self._lease_ttl_s
+        for key in (source_key, _GLOBAL):
+            self._rpm[(key, window)] = self._rpm.get((key, window), 0) + 1
+            if est_tokens:
+                self._tpm[(key, window)] = self._tpm.get((key, window), 0) + est_tokens
+        return _MemoryPermit(self, source_key, lease_id, est_tokens, window)
+
+    async def acquire(self, source_key: str, est_tokens: int) -> _MemoryPermit:
+        """阻塞准入: 轮询直到拿到 permit;sleep 可被取消(取消穿透铁律)。"""
+        while True:
+            permit = await self.try_acquire(source_key, est_tokens)
+            if permit is not None:
+                return permit
+            await self._sleep(self._poll_interval_s)
+
+    async def source_stats(self, source_key: str) -> SourceStats:
+        self._cfg(source_key)
+        self._purge_leases()
+        window = self._window()
+        return SourceStats(
+            inflight=self._inflight(source_key),
+            rpm_used=self._rpm.get((source_key, window), 0),
+            tpm_used=self._tpm.get((source_key, window), 0),
+        )
+
+    async def mark_progress(self) -> None:
+        """记录"最近一次出餐"时刻,供背压 stall 判定(M2)读取。"""
+        self._progress_at = self._now()
+
+    async def progress_age_s(self) -> float:
+        if self._progress_at is None:
+            return float("inf")
+        return self._now() - self._progress_at
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/contracts/conftest.py b/tests/contracts/conftest.py
new file mode 100644
index 0000000..de6e87b
--- /dev/null
+++ b/tests/contracts/conftest.py
@@ -0,0 +1,61 @@
+"""契约测试共享 fixture: 后端参数化(M1 仅 memory,M2 增 redis 零改测试)。
+
+FakeClock 仅对支持时钟注入的后端有效(memory);M2 接入 Redis 后端时,
+依赖时钟推进的用例按后端能力跳过或改用真实等待。
+"""
+
+import pytest
+
+from polygateway.backends.memory.breaker import InMemoryGate
+from polygateway.backends.memory.limiter import InMemoryLimiter
+from polygateway.types import BreakerConfig, GlobalLimits, SourceConfig
+
+
+class FakeClock:
+    """确定性单调时钟;契约测试推进时间验证租约/冷却语义。"""
+
+    def __init__(self, start: float = 1000.0) -> None:
+        self.t = start
+
+    def __call__(self) -> float:
+        return self.t
+
+    def advance(self, seconds: float) -> None:
+        self.t += seconds
+
+
+def make_source(name: str = "s1", **overrides) -> SourceConfig:
+    base = dict(
+        name=name, provider="openai", base_url="https://gw.example/v1",
+        api_key="sk-test", model="m", timeout_s=10.0,
+    )
+    base.update(overrides)
+    return SourceConfig(**base)
+
+
+@pytest.fixture
+def clock() -> FakeClock:
+    return FakeClock()
+
+
+@pytest.fixture(params=["memory"])
+def limiter_factory(request, clock):
+    """返回 (sources, global_limits, lease_ttl_s) -> RateLimiter 的工厂。"""
+
+    def make(sources: list[SourceConfig], global_limits: GlobalLimits, lease_ttl_s: float = 100.0):
+        return InMemoryLimiter(
+            scope="llm", sources={s.name: s for s in sources},
+            global_limits=global_limits, lease_ttl_s=lease_ttl_s, now=clock,
+        )
+
+    return make
+
+
+@pytest.fixture(params=["memory"])
+def gate_factory(request, clock):
+    """返回 (BreakerConfig) -> ProviderGate 的工厂。"""
+
+    def make(config: BreakerConfig):
+        return InMemoryGate(config=config, now=clock)
+
+    return make
diff --git a/tests/contracts/test_breaker_contract.py b/tests/contracts/test_breaker_contract.py
new file mode 100644
index 0000000..c9624d9
--- /dev/null
+++ b/tests/contracts/test_breaker_contract.py
@@ -0,0 +1,151 @@
+"""熔断后端契约测试(状态机 + 半开单探针租约 + epoch fencing)。
+
+蓝本: VT adapters/breaker.py 状态机语义 + CHS provider_gate.py 契约;
+M2 的 Redis 实现复用本套件。
+"""
+
+import pytest
+
+from polygateway.ports import GateState
+from polygateway.types import BreakerConfig
+
+_CFG = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
+
+
+async def _open_gate(gate, source="s1"):
+    """连续失败到阈值,打开熔断,返回最后一次 Update。"""
+    update = None
+    for _ in range(_CFG.fail_threshold):
+        entry = await gate.try_enter(source, "w")
+        assert entry.allowed
+        update = await gate.record_failure(entry, "network_error", False)
+    assert update.state is GateState.OPEN
+    return update
+
+
+class TestStateMachine:
+    async def test_closed_admits_and_success_resets(self, gate_factory):
+        gate = gate_factory(_CFG)
+        entry = await gate.try_enter("s1", "w")
+        assert entry.allowed and entry.state is GateState.CLOSED and not entry.is_probe
+        update = await gate.record_success(entry)
+        assert update.applied and update.failure_count == 0
+
+    async def test_threshold_opens_and_rejects(self, gate_factory):
+        gate = gate_factory(_CFG)
+        await _open_gate(gate)
+        entry = await gate.try_enter("s1", "w")
+        assert not entry.allowed and entry.state is GateState.OPEN
+        assert entry.retry_after_s > 0
+
+    async def test_success_before_threshold_resets_count(self, gate_factory):
+        gate = gate_factory(_CFG)
+        for _ in range(_CFG.fail_threshold - 1):
+            entry = await gate.try_enter("s1", "w")
+            await gate.record_failure(entry, "timeout", False)
+        entry = await gate.try_enter("s1", "w")
+        update = await gate.record_success(entry)
+        assert update.applied and update.failure_count == 0
+        assert (await gate.try_enter("s1", "w")).allowed
+
+    async def test_force_open_single_strike(self, gate_factory):
+        gate = gate_factory(_CFG)
+        entry = await gate.try_enter("s1", "w")
+        update = await gate.record_failure(entry, "source_dead", True)
+        assert update.state is GateState.OPEN
+        assert not (await gate.try_enter("s1", "w")).allowed
+
+
+class TestHalfOpenProbe:
+    async def test_cooldown_grants_single_probe(self, gate_factory, clock):
+        gate = gate_factory(_CFG)
+        await _open_gate(gate)
+        clock.advance(_CFG.cooldown_s + 1)
+        probe = await gate.try_enter("s1", "w1")
+        assert probe.allowed and probe.is_probe and probe.probe_owner == "w1"
+        # 第二个进入者被拒(防惊群)
+        second = await gate.try_enter("s1", "w2")
+        assert not second.allowed and second.state is GateState.HALF_OPEN
+
+    async def test_probe_success_closes(self, gate_factory, clock):
+        gate = gate_factory(_CFG)
+        await _open_gate(gate)
+        clock.advance(_CFG.cooldown_s + 1)
+        probe = await gate.try_enter("s1", "w1")
+        update = await gate.record_success(probe)
+        assert update.applied and update.state is GateState.CLOSED
+        assert (await gate.try_enter("s1", "w2")).allowed
+
+    async def test_probe_failure_reopens(self, gate_factory, clock):
+        gate = gate_factory(_CFG)
+        await _open_gate(gate)
+        clock.advance(_CFG.cooldown_s + 1)
+        probe = await gate.try_enter("s1", "w1")
+        update = await gate.record_failure(probe, "network_error", False)
+        assert update.state is GateState.OPEN
+        assert not (await gate.try_enter("s1", "w2")).allowed
+
+    async def test_probe_lease_expiry_allows_takeover(self, gate_factory, clock):
+        """探针持有者死亡 → 租约过期后新 caller 接管探针,防死锁。"""
+        gate = gate_factory(_CFG)
+        await _open_gate(gate)
+        clock.advance(_CFG.cooldown_s + 1)
+        stale = await gate.try_enter("s1", "dead-worker")
+        assert stale.is_probe
+        clock.advance(_CFG.probe_ttl_s + 1)
+        takeover = await gate.try_enter("s1", "w2")
+        assert takeover.allowed and takeover.is_probe and takeover.probe_owner == "w2"
+
+    async def test_release_probe_hands_back_and_idempotent(self, gate_factory, clock):
+        gate = gate_factory(_CFG)
+        await _open_gate(gate)
+        clock.advance(_CFG.cooldown_s + 1)
+        probe = await gate.try_enter("s1", "w1")
+        update = await gate.release_probe(probe)
+        assert update.applied
+        # 幂等: 第二次释放不再生效
+        assert not (await gate.release_probe(probe)).applied
+        # 源保持可接管状态: 下一 caller 拿到探针
+        nxt = await gate.try_enter("s1", "w2")
+        assert nxt.allowed and nxt.is_probe
+
+
+class TestEpochFencing:
+    async def test_stale_epoch_write_rejected(self, gate_factory, clock):
+        """entry 取得后世代已推进(他人触发开路)→ 迟到写回被 fencing 拒绝。"""
+        gate = gate_factory(_CFG)
+        stale_entry = await gate.try_enter("s1", "slow-worker")
+        await _open_gate(gate)  # 他人连续失败 → 开路,epoch 推进
+        late = await gate.record_success(stale_entry)
+        assert not late.applied
+        # 门仍是 OPEN,未被迟到的成功污染
+        assert not (await gate.try_enter("s1", "w")).allowed
+
+    async def test_stale_probe_owner_rejected(self, gate_factory, clock):
+        gate = gate_factory(_CFG)
+        await _open_gate(gate)
+        clock.advance(_CFG.cooldown_s + 1)
+        stale_probe = await gate.try_enter("s1", "dead-worker")
+        clock.advance(_CFG.probe_ttl_s + 1)
+        takeover = await gate.try_enter("s1", "w2")
+        assert takeover.is_probe
+        # 死亡探针的迟到写回被拒(owner 已易主)
+        assert not (await gate.record_success(stale_probe)).applied
+
+
+class TestRetryAfter:
+    async def test_retry_after_semantics(self, gate_factory, clock):
+        gate = gate_factory(_CFG)
+        assert await gate.retry_after_s(("s1",)) == 0.0  # 健康 → 0
+        await _open_gate(gate)
+        wait = await gate.retry_after_s(("s1",))
+        assert 0 < wait <= _CFG.cooldown_s
+        clock.advance(_CFG.cooldown_s + 1)
+        assert await gate.retry_after_s(("s1",)) == 0.0  # 冷却到期 → 0
+        with pytest.raises(ValueError):
+            await gate.retry_after_s(())
+
+    async def test_retry_after_takes_min_across_sources(self, gate_factory, clock):
+        gate = gate_factory(_CFG)
+        await _open_gate(gate, "s1")  # s1 开路;s2 健康
+        assert await gate.retry_after_s(("s1", "s2")) == 0.0
diff --git a/tests/contracts/test_limiter_contract.py b/tests/contracts/test_limiter_contract.py
new file mode 100644
index 0000000..bc102cb
--- /dev/null
+++ b/tests/contracts/test_limiter_contract.py
@@ -0,0 +1,103 @@
+"""限流后端契约测试(CHS tests/contracts_limiter.py 5 项 + M1 设计 §4.2 补强)。
+
+任何 RateLimiter 后端都必须逐条通过;M2 的 Redis 实现复用本套件。
+"""
+
+from polygateway.types import GlobalLimits
+from tests.contracts.conftest import make_source
+
+_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
+
+
+class TestConcurrencyGate:
+    async def test_concurrency_caps_and_zero_side_effect(self, limiter_factory):
+        src = make_source(max_concurrency=2)
+        limiter = limiter_factory([src], _NO_GLOBAL)
+        p1 = await limiter.try_acquire("s1", 0)
+        p2 = await limiter.try_acquire("s1", 0)
+        assert p1 is not None and p2 is not None
+        # 满员 → None,且失败的 acquire 零副作用
+        assert await limiter.try_acquire("s1", 0) is None
+        stats = await limiter.source_stats("s1")
+        assert stats.inflight == 2
+        # 释放一个后又能进
+        await p1.release()
+        assert (await limiter.source_stats("s1")).inflight == 1
+        p3 = await limiter.try_acquire("s1", 0)
+        assert p3 is not None
+        await p2.release()
+        await p3.release()
+        assert (await limiter.source_stats("s1")).inflight == 0
+
+    async def test_release_idempotent(self, limiter_factory):
+        limiter = limiter_factory([make_source(max_concurrency=1)], _NO_GLOBAL)
+        permit = await limiter.try_acquire("s1", 0)
+        await permit.release()
+        await permit.release()
+        assert (await limiter.source_stats("s1")).inflight == 0
+
+    async def test_global_concurrency_across_sources(self, limiter_factory):
+        sources = [make_source("s1"), make_source("s2")]
+        limiter = limiter_factory(sources, GlobalLimits(max_concurrency=2, rpm=0, tpm=0))
+        assert await limiter.try_acquire("s1", 0) is not None
+        assert await limiter.try_acquire("s2", 0) is not None
+        assert await limiter.try_acquire("s1", 0) is None  # 全局闸挡住第三个
+
+    async def test_lease_expiry_reclaims_slot(self, limiter_factory, clock):
+        """permit 持有者死亡(未 release)→ 租约过期后并发槽自动回收。"""
+        limiter = limiter_factory([make_source(max_concurrency=1)], _NO_GLOBAL, lease_ttl_s=30.0)
+        _leaked = await limiter.try_acquire("s1", 0)
+        assert await limiter.try_acquire("s1", 0) is None
+        clock.advance(31.0)
+        assert await limiter.try_acquire("s1", 0) is not None
+
+
+class TestRpmGate:
+    async def test_rpm_not_refunded_by_release(self, limiter_factory):
+        src = make_source(rpm=3)
+        limiter = limiter_factory([src], _NO_GLOBAL)
+        for _ in range(3):
+            permit = await limiter.try_acquire("s1", 0)
+            assert permit is not None
+            await permit.release()  # 释放并发,但 RPM 计数不归还
+        assert await limiter.try_acquire("s1", 0) is None
+        assert (await limiter.source_stats("s1")).rpm_used == 3
+
+
+class TestTpmGate:
+    async def test_prededuct_and_settle_refund(self, limiter_factory):
+        src = make_source(tpm=1000, est_tokens=400)
+        limiter = limiter_factory([src], _NO_GLOBAL)
+        p1 = await limiter.try_acquire("s1", 400)
+        p2 = await limiter.try_acquire("s1", 400)
+        assert p1 is not None and p2 is not None
+        assert await limiter.try_acquire("s1", 400) is None  # 预扣用满
+        # 实际 0 tokens → 全额退款
+        await p1.settle(0)
+        await p1.release()
+        assert (await limiter.source_stats("s1")).tpm_used == 400
+        # settle 幂等: 第二次调用无副作用
+        await p1.settle(0)
+        assert (await limiter.source_stats("s1")).tpm_used == 400
+        # 多退少补: 实际超预扣则补记
+        await p2.settle(600)
+        await p2.release()
+        assert (await limiter.source_stats("s1")).tpm_used == 600
+
+    async def test_failed_acquire_leaves_no_tpm_trace(self, limiter_factory):
+        src = make_source(tpm=500, est_tokens=400)
+        limiter = limiter_factory([src], _NO_GLOBAL)
+        p1 = await limiter.try_acquire("s1", 400)
+        assert p1 is not None
+        assert await limiter.try_acquire("s1", 400) is None
+        assert (await limiter.source_stats("s1")).tpm_used == 400  # 失败尝试零痕迹
+
+
+class TestProgress:
+    async def test_progress_marks_fresh(self, limiter_factory, clock):
+        limiter = limiter_factory([make_source()], _NO_GLOBAL)
+        assert await limiter.progress_age_s() == float("inf")  # 从未出餐
+        await limiter.mark_progress()
+        assert await limiter.progress_age_s() < 5.0
+        clock.advance(42.0)
+        assert 41.0 < await limiter.progress_age_s() < 43.0
diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py
new file mode 100644
index 0000000..e69de29