iomgaa
f958138e83
Telemetry degradation used to be a single warning and a private boolean. In a long-running process that is indistinguishable from telemetry working: issue #15 was only found by hand-reconciling milestone log lines against llm_calls rows, after 19 calls had silently gone unrecorded. The SQLite side was worse — once init failed, every write returned without even a log line. Degradation now has one shared owner. TelemetryStatusTracker holds the state machine (enter/recover/drop/should-retry), announces entry and recovery once each, and repeats the drop count under a row-and-time double threshold so a degraded backend neither floods the log nor goes quiet. Both recorders hold one; both count the rows they drop. For programmatic consumers, TelemetryStatus is a frozen snapshot exposed as telemetry_status on all three clients, resolved through a single isinstance check. It is a separate optional port rather than a member of TelemetryRecorder: that protocol is @runtime_checkable, so adding an attribute would make every implementation that only defines record_llm_call stop satisfying it — downstream isinstance assertions would break on upgrade. The existing assertion in test_ports.py is what keeps that decision honest. Failure criteria are deliberately untouched here: Postgres still treats a pool failure as permanent, only now visibly. `_failed` and the tracker therefore both carry the verdict for the span of this one change; the cooldown rework collapses them into the tracker alone.
283 lines
8.3 KiB
Python
283 lines
8.3 KiB
Python
"""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,
|
|
TelemetryStatusProvider,
|
|
Transport,
|
|
)
|
|
from polygateway.types import LLMResponse, SourceStats, TelemetryStatus
|
|
|
|
|
|
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,
|
|
cached_prompt_tokens,
|
|
model_reported,
|
|
sampling,
|
|
reasoning_tokens,
|
|
tenant_id,
|
|
meta,
|
|
) -> 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)
|
|
|
|
|
|
class _DummyStatusProvider(_DummyRecorder):
|
|
@property
|
|
def telemetry_status(self) -> TelemetryStatus:
|
|
return TelemetryStatus(
|
|
degraded=False,
|
|
fatal=False,
|
|
reason=None,
|
|
degraded_for_s=None,
|
|
dropped_rows=0,
|
|
retry_after_s=None,
|
|
)
|
|
|
|
|
|
def test_status_provider_is_a_separate_optional_port():
|
|
"""状态**不得**并进 TelemetryRecorder: 那会让只实现 record_llm_call 的对象
|
|
当场不再满足 @runtime_checkable 的结构检查(设计 §3.3,Codex 审查)。"""
|
|
assert isinstance(_DummyStatusProvider(), TelemetryStatusProvider)
|
|
assert isinstance(_DummyStatusProvider(), TelemetryRecorder)
|
|
assert not isinstance(_DummyRecorder(), TelemetryStatusProvider)
|
|
assert isinstance(_DummyRecorder(), TelemetryRecorder) # 这条断言是那条决策的执法点
|
|
|
|
|
|
def _decision(**overrides) -> GateDecision:
|
|
base = {
|
|
"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
|
|
)
|
|
|
|
|
|
class TestTelemetryRecorderSignature:
|
|
"""`record_llm_call` 的冻结签名以 `inspect.signature` 实测,不凭记忆断言。
|
|
|
|
该 Protocol 的纪律是新增参数**不设默认值**(ports.py docstring):库外无第三方
|
|
实现者,而带默认值的参数会让 emitter 漏传时静默落默认值——遥测里的租户归属
|
|
一旦静默错位,事后无从分辨是"没传"还是"就是空的"。
|
|
"""
|
|
|
|
def test_caller_dimensions_are_declared(self):
|
|
import inspect
|
|
|
|
params = inspect.signature(TelemetryRecorder.record_llm_call).parameters
|
|
assert {"tenant_id", "meta"} <= set(params)
|
|
|
|
@pytest.mark.parametrize("name", ["tenant_id", "meta"])
|
|
def test_caller_dimensions_have_no_default(self, name):
|
|
import inspect
|
|
|
|
param = inspect.signature(TelemetryRecorder.record_llm_call).parameters[name]
|
|
assert param.default is inspect.Parameter.empty
|
|
assert param.kind is inspect.Parameter.KEYWORD_ONLY
|
|
|
|
|
|
class TestOcrPorts:
|
|
"""M3 三个 OCR Protocol(设计 §3.2): runtime_checkable 结构判定。"""
|
|
|
|
def test_ocr_ports_runtime_checkable(self):
|
|
from polygateway.ports import OcrLayoutPort, OcrTextPort, OcrTransport
|
|
|
|
class GoodClient:
|
|
async def recognize_text(self, image): ...
|
|
async def parse_layout(self, image): ...
|
|
|
|
class GoodTransport:
|
|
async def recognize_text(self, *, image, source, call_id): ...
|
|
async def parse_layout(self, *, image, source, call_id): ...
|
|
async def check_health(self, *, source): ...
|
|
|
|
assert isinstance(GoodClient(), OcrTextPort)
|
|
assert isinstance(GoodClient(), OcrLayoutPort)
|
|
assert isinstance(GoodTransport(), OcrTransport)
|
|
|
|
def test_missing_method_rejected(self):
|
|
from polygateway.ports import OcrTransport
|
|
|
|
class NoHealth:
|
|
async def recognize_text(self, *, image, source, call_id): ...
|
|
async def parse_layout(self, *, image, source, call_id): ...
|
|
|
|
assert not isinstance(NoHealth(), OcrTransport)
|