feat: freeze all port protocols and gate snapshots
This commit is contained in:
@@ -0,0 +1,202 @@
|
||||
"""全部端口协议与治理快照类型(M1 设计 §4;最内层,只依赖 types/errors/标准库)。
|
||||
|
||||
决策逻辑与状态存储分离(D3): 中间件只面向这里的 Protocol,
|
||||
后端(memory/redis)各自实现同一契约并共用一套契约测试。
|
||||
时间量纲一律**秒**(CHS Redis 实现内部的毫秒换算是后端私事,不进契约)。
|
||||
"""
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
from dataclasses import dataclass
|
||||
from enum import StrEnum
|
||||
from typing import Any, Protocol, runtime_checkable
|
||||
|
||||
from .types import ChatRequest, LLMResponse, SourceConfig, SourceStats, TransportResult
|
||||
|
||||
CallNext = Callable[[ChatRequest], Awaitable[LLMResponse]]
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class Middleware(Protocol):
|
||||
"""洋葱层: 包裹 call_next,层与层正交(D1)。"""
|
||||
|
||||
async def __call__(self, request: ChatRequest, call_next: CallNext) -> LLMResponse: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class Transport(Protocol):
|
||||
"""一次原始调用的协议细节(请求组装/流式解析/错误翻译);不含任何治理。"""
|
||||
|
||||
async def complete(
|
||||
self,
|
||||
*,
|
||||
messages: list[dict[str, Any]],
|
||||
source: SourceConfig,
|
||||
stream: bool,
|
||||
overlay: dict[str, Any],
|
||||
call_id: str,
|
||||
) -> TransportResult: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class Permit(Protocol):
|
||||
"""限流入场许可;settle/release 均幂等,finally 中必然执行。"""
|
||||
|
||||
async def release(self) -> None: ...
|
||||
|
||||
async def settle(self, actual_tokens: int) -> None: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class RateLimiter(Protocol):
|
||||
"""限流契约(CHS limiter.py 形态): 并发/RPM/TPM × 全局/单源 六道闸。"""
|
||||
|
||||
async def try_acquire(self, source_key: str, est_tokens: int) -> Permit | None: ...
|
||||
|
||||
async def acquire(self, source_key: str, est_tokens: int) -> Permit: ...
|
||||
|
||||
async def source_stats(self, source_key: str) -> SourceStats: ...
|
||||
|
||||
async def mark_progress(self) -> None: ...
|
||||
|
||||
async def progress_age_s(self) -> float: ...
|
||||
|
||||
|
||||
class GateState(StrEnum):
|
||||
"""熔断状态机三态。"""
|
||||
|
||||
CLOSED = "closed"
|
||||
OPEN = "open"
|
||||
HALF_OPEN = "half_open"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GateDecision:
|
||||
"""一次源准入决定,同时是结果写回所需的 fencing token(CHS ports.py:405)。"""
|
||||
|
||||
source_name: str
|
||||
allowed: bool
|
||||
state: GateState
|
||||
epoch: int
|
||||
is_probe: bool
|
||||
probe_owner: str | None
|
||||
retry_after_s: float
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
# 校验逐条移植 CHS ports.py:405-440,保证快照内部一致
|
||||
self._check_bounds()
|
||||
self._check_admission()
|
||||
self._check_probe()
|
||||
|
||||
def _check_bounds(self) -> None:
|
||||
if not self.source_name.strip():
|
||||
raise ValueError("source_name 不能为空")
|
||||
if self.epoch < 0:
|
||||
raise ValueError("epoch 不能为负")
|
||||
if self.retry_after_s < 0:
|
||||
raise ValueError("retry_after_s 不能为负")
|
||||
|
||||
def _check_admission(self) -> None:
|
||||
if self.allowed and self.state is GateState.OPEN:
|
||||
raise ValueError("OPEN 状态不得准入")
|
||||
if self.allowed and self.state is GateState.HALF_OPEN and not self.is_probe:
|
||||
raise ValueError("HALF_OPEN 准入必须是探针")
|
||||
|
||||
def _check_probe(self) -> None:
|
||||
if self.is_probe:
|
||||
probe_valid = (
|
||||
self.allowed
|
||||
and self.state is GateState.HALF_OPEN
|
||||
and self.probe_owner is not None
|
||||
and bool(self.probe_owner.strip())
|
||||
)
|
||||
if not probe_valid:
|
||||
raise ValueError("探针决定必须同时包含准入、HALF_OPEN 与 owner")
|
||||
elif self.probe_owner is not None:
|
||||
raise ValueError("非探针决定不得携带 probe_owner")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GateUpdate:
|
||||
"""门控写回快照;applied=False 表示 fencing 已拒绝迟到结果(CHS ports.py:442)。"""
|
||||
|
||||
applied: bool
|
||||
state: GateState
|
||||
epoch: int
|
||||
failure_count: int
|
||||
retry_after_s: float
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.epoch < 0 or self.failure_count < 0 or self.retry_after_s < 0:
|
||||
raise ValueError("门控更新快照不能包含负数")
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ProviderGate(Protocol):
|
||||
"""熔断契约(CHS provider_gate.py 形态): 半开单探针租约 + epoch fencing 在契约内。"""
|
||||
|
||||
async def try_enter(self, source_name: str, owner: str) -> GateDecision: ...
|
||||
|
||||
async def record_success(self, entry: GateDecision) -> GateUpdate: ...
|
||||
|
||||
async def record_failure(
|
||||
self, entry: GateDecision, reason: str, force_open: bool
|
||||
) -> GateUpdate: ...
|
||||
|
||||
async def release_probe(self, entry: GateDecision) -> GateUpdate: ...
|
||||
|
||||
async def retry_after_s(self, sources: tuple[str, ...]) -> float: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class CacheBackend(Protocol):
|
||||
"""响应缓存后端: 笨 KV;key 公式归 CacheMW(算法一份)。"""
|
||||
|
||||
async def get(self, key: str) -> str | None: ...
|
||||
|
||||
async def set(self, key: str, value: str, ttl_s: int) -> None: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class SourceSelector(Protocol):
|
||||
"""选源策略(CHS selector.py 形态): 返回本次尝试的候选顺序。"""
|
||||
|
||||
def order(
|
||||
self, sources: list[SourceConfig], stats: dict[str, SourceStats]
|
||||
) -> list[SourceConfig]: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class StructuredOutputStrategy(Protocol):
|
||||
"""结构化输出策略(D7/D14): 请求侧叠加 + 响应侧解析。"""
|
||||
|
||||
def request_overlay(self, schema: dict[str, Any] | None) -> dict[str, Any]: ...
|
||||
|
||||
def parse(self, text: str) -> Any: ...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class TelemetryRecorder(Protocol):
|
||||
"""遥测后端;18 字段冻结(M1 设计 §4.4),唯一调用点是 TelemetryEmitter。"""
|
||||
|
||||
async def record_llm_call(
|
||||
self,
|
||||
*,
|
||||
call_id: str,
|
||||
parent_call_id: str | None,
|
||||
session_id: str | None,
|
||||
model: str,
|
||||
provider: str,
|
||||
source_name: str,
|
||||
messages: str,
|
||||
response: str,
|
||||
thinking: str,
|
||||
prompt_tokens: int,
|
||||
completion_tokens: int,
|
||||
usage_source: str,
|
||||
latency_ms: int,
|
||||
ttft_ms: float | None,
|
||||
max_inter_token_ms: float | None,
|
||||
cache_hit: bool,
|
||||
error: str | None,
|
||||
cost: float | None,
|
||||
) -> None: ...
|
||||
@@ -0,0 +1,149 @@
|
||||
"""ports.py 端口冻结测试(M1 设计 §4): Protocol 结构性检查 + Gate 快照校验。"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from polygateway.ports import (
|
||||
CacheBackend,
|
||||
GateDecision,
|
||||
GateState,
|
||||
GateUpdate,
|
||||
Middleware,
|
||||
Permit,
|
||||
ProviderGate,
|
||||
RateLimiter,
|
||||
SourceSelector,
|
||||
StructuredOutputStrategy,
|
||||
TelemetryRecorder,
|
||||
Transport,
|
||||
)
|
||||
from polygateway.types import LLMResponse, SourceStats
|
||||
|
||||
|
||||
def _resp() -> LLMResponse:
|
||||
return LLMResponse("c", "t", "m", "p", 1, 2, 3, None, None, False, "cid")
|
||||
|
||||
|
||||
class _DummyPermit:
|
||||
async def release(self) -> None: ...
|
||||
async def settle(self, actual_tokens: int) -> None: ...
|
||||
|
||||
|
||||
class _DummyLimiter:
|
||||
async def try_acquire(self, source_key: str, est_tokens: int): return _DummyPermit()
|
||||
async def acquire(self, source_key: str, est_tokens: int): return _DummyPermit()
|
||||
async def source_stats(self, source_key: str): return SourceStats(0, 0, 0)
|
||||
async def mark_progress(self) -> None: ...
|
||||
async def progress_age_s(self) -> float: return 0.0
|
||||
|
||||
|
||||
class _DummyGate:
|
||||
async def try_enter(self, source_name: str, owner: str): raise NotImplementedError
|
||||
async def record_success(self, entry): raise NotImplementedError
|
||||
async def record_failure(self, entry, reason: str, force_open: bool): raise NotImplementedError
|
||||
async def release_probe(self, entry): raise NotImplementedError
|
||||
async def retry_after_s(self, sources): return 0.0
|
||||
|
||||
|
||||
class _DummyMw:
|
||||
async def __call__(self, request, call_next): return await call_next(request)
|
||||
|
||||
|
||||
class _DummyTransport:
|
||||
async def complete(self, *, messages, source, stream, overlay, call_id): raise NotImplementedError
|
||||
|
||||
|
||||
class _DummyCache:
|
||||
async def get(self, key: str): return None
|
||||
async def set(self, key: str, value: str, ttl_s: int) -> None: ...
|
||||
|
||||
|
||||
class _DummySelector:
|
||||
def order(self, sources, stats): return list(sources)
|
||||
|
||||
|
||||
class _DummyStrategy:
|
||||
def request_overlay(self, schema): return {}
|
||||
def parse(self, text: str) -> Any: return {}
|
||||
|
||||
|
||||
class _DummyRecorder:
|
||||
async def record_llm_call(
|
||||
self, *, call_id, parent_call_id, session_id, model, provider, source_name,
|
||||
messages, response, thinking, prompt_tokens, completion_tokens, usage_source,
|
||||
latency_ms, ttft_ms, max_inter_token_ms, cache_hit, error, cost,
|
||||
) -> None: ...
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("impl", "protocol"),
|
||||
[
|
||||
(_DummyPermit(), Permit),
|
||||
(_DummyLimiter(), RateLimiter),
|
||||
(_DummyGate(), ProviderGate),
|
||||
(_DummyMw(), Middleware),
|
||||
(_DummyTransport(), Transport),
|
||||
(_DummyCache(), CacheBackend),
|
||||
(_DummySelector(), SourceSelector),
|
||||
(_DummyStrategy(), StructuredOutputStrategy),
|
||||
(_DummyRecorder(), TelemetryRecorder),
|
||||
],
|
||||
)
|
||||
def test_protocols_are_runtime_checkable(impl, protocol):
|
||||
assert isinstance(impl, protocol)
|
||||
|
||||
|
||||
def _decision(**overrides) -> GateDecision:
|
||||
base = dict(
|
||||
source_name="qwen_1", allowed=True, state=GateState.CLOSED,
|
||||
epoch=0, is_probe=False, probe_owner=None, retry_after_s=0.0,
|
||||
)
|
||||
base.update(overrides)
|
||||
return GateDecision(**base)
|
||||
|
||||
|
||||
class TestGateDecisionInvariants:
|
||||
"""校验逐条移植 CHS ports.py:405-440。"""
|
||||
|
||||
def test_valid_probe_decision(self):
|
||||
d = _decision(state=GateState.HALF_OPEN, is_probe=True, probe_owner="w1")
|
||||
assert d.is_probe and d.probe_owner == "w1"
|
||||
|
||||
def test_empty_source_rejected(self):
|
||||
with pytest.raises(ValueError):
|
||||
_decision(source_name=" ")
|
||||
|
||||
def test_negative_epoch_and_retry_after_rejected(self):
|
||||
with pytest.raises(ValueError):
|
||||
_decision(epoch=-1)
|
||||
with pytest.raises(ValueError):
|
||||
_decision(retry_after_s=-0.1)
|
||||
|
||||
def test_open_state_cannot_allow(self):
|
||||
with pytest.raises(ValueError):
|
||||
_decision(state=GateState.OPEN, allowed=True)
|
||||
|
||||
def test_half_open_admission_must_be_probe(self):
|
||||
with pytest.raises(ValueError):
|
||||
_decision(state=GateState.HALF_OPEN, is_probe=False)
|
||||
|
||||
def test_probe_requires_owner_and_half_open(self):
|
||||
with pytest.raises(ValueError):
|
||||
_decision(state=GateState.HALF_OPEN, is_probe=True, probe_owner=None)
|
||||
with pytest.raises(ValueError):
|
||||
_decision(state=GateState.CLOSED, is_probe=True, probe_owner="w1")
|
||||
|
||||
def test_non_probe_cannot_carry_owner(self):
|
||||
with pytest.raises(ValueError):
|
||||
_decision(probe_owner="w1")
|
||||
|
||||
|
||||
class TestGateUpdate:
|
||||
def test_bounds(self):
|
||||
u = GateUpdate(applied=True, state=GateState.CLOSED, epoch=0, failure_count=0, retry_after_s=0.0)
|
||||
assert u.applied
|
||||
with pytest.raises(ValueError):
|
||||
GateUpdate(applied=True, state=GateState.CLOSED, epoch=-1, failure_count=0, retry_after_s=0.0)
|
||||
with pytest.raises(ValueError):
|
||||
GateUpdate(applied=True, state=GateState.CLOSED, epoch=0, failure_count=-1, retry_after_s=0.0)
|
||||
Reference in New Issue
Block a user