diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py new file mode 100644 index 0000000..c66180f --- /dev/null +++ b/src/polygateway/ports.py @@ -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: ... diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py new file mode 100644 index 0000000..730a3cc --- /dev/null +++ b/tests/unit/test_ports.py @@ -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)