iomgaa
393f2bf617
Grow the telemetry contract from 26 to 36 fields and give every logical call a failure terminal row, so SQL can finally answer "how many calls failed" and "why did the whole pool die". Schema and port move together with the emitter writes in one commit: splitting them would ship columns that nothing populates. - schema: append 10 nullable columns (scope, operation, logical_call_id, event_kind, http_status_code, error_type, cause_type, error_body, attempts, total_latency_ms) to all five definition sites in one order - ports: 10 keyword-only parameters without defaults; the protocol signature is now the single source the assembly gate derives from - emitter: take domain exception objects instead of pre-flattened text and pin down the diagnostics in one helper; a relabelled 503 stays 503 and success rows leave all five columns NULL - emitter: reject recorders whose record_llm_call cannot accept the current field shape at assembly time, since _record would otherwise swallow the TypeError and drop every row while calls keep succeeding - clients: write at most one terminal row per logical call through a single shared exit, deduplicated by the call context; TelemetryMW stops writing terminals so the two sites cannot double count - clients: cancellation stays best effort and propagates, non-domain exceptions get no terminal row and keep their classification - transports: give _status_to_error an explicit operation and fix the historically mislabelled embedding HTTP failures - structured: promote the bounded error formatter so the reask feedback and the terminal explanation share one set of limits Terminal rows carry no cost and no tokens, so cost aggregation is unchanged; failure counts must now filter on event_kind.
367 lines
11 KiB
Python
367 lines
11 KiB
Python
"""ports.py 端口冻结测试(M1 设计 §4): Protocol 结构性检查 + Gate 快照校验。"""
|
|
|
|
import inspect
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from polygateway.ports import (
|
|
CacheBackend,
|
|
EmbeddingTransport,
|
|
GateDecision,
|
|
GateState,
|
|
GateUpdate,
|
|
Middleware,
|
|
OcrTransport,
|
|
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, reasoning_effort):
|
|
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,
|
|
thinking_observation,
|
|
reasoning_effort,
|
|
scope,
|
|
operation,
|
|
logical_call_id,
|
|
event_kind,
|
|
http_status_code,
|
|
error_type,
|
|
cause_type,
|
|
error_body,
|
|
attempts,
|
|
total_latency_ms,
|
|
) -> 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 TestReasoningTierIsOnlyOnTheChatPort:
|
|
"""档位属于 chat 端口,且**只属于**它(Task 5b)。
|
|
|
|
`@runtime_checkable` 只查方法名不查签名,故协议签名本身必须被显式断言——
|
|
否则实现漏改一个参数,要到运行期调用才会以 `TypeError` 现形,而那时的现场
|
|
离根因已经很远。
|
|
"""
|
|
|
|
def test_chat_transport_carries_the_per_call_tier(self):
|
|
params = inspect.signature(Transport.complete).parameters
|
|
assert "reasoning_effort" in params
|
|
# 不给默认值是有意的(与 TelemetryRecorder 同一既有约定): 库外无第三方
|
|
# 实现者,写全签名成本为零,而默认值会把"漏传"变成静默的"不表态"
|
|
assert params["reasoning_effort"].default is inspect.Parameter.empty
|
|
|
|
@pytest.mark.parametrize(
|
|
("protocol", "method"),
|
|
[
|
|
(EmbeddingTransport, "embed"),
|
|
(OcrTransport, "recognize_text"),
|
|
(OcrTransport, "parse_layout"),
|
|
],
|
|
)
|
|
def test_other_transports_have_no_reasoning_tier(self, protocol, method):
|
|
"""embedding 与 OCR 没有推理语义,给它们加档位只会静默无效(issue #4 同款决策)。"""
|
|
assert "reasoning_effort" not in inspect.signature(getattr(protocol, method)).parameters
|
|
|
|
|
|
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)
|
|
|
|
def test_call_observability_fields_are_declared(self):
|
|
"""1.3.5 十列进协议(issue #19/#23);字段总数以实测为准不凭记忆。
|
|
|
|
本签名同时是装配闸的事实源(`_assert_recorder_shape` 按它派生参数名),
|
|
故它与实现一旦漂移,下游自定义 recorder 会在装配期就被拒。
|
|
"""
|
|
import inspect
|
|
|
|
params = inspect.signature(TelemetryRecorder.record_llm_call).parameters
|
|
assert {
|
|
"scope",
|
|
"operation",
|
|
"logical_call_id",
|
|
"event_kind",
|
|
"http_status_code",
|
|
"error_type",
|
|
"cause_type",
|
|
"error_body",
|
|
"attempts",
|
|
"total_latency_ms",
|
|
} <= set(params)
|
|
assert len(params) - 1 == 36 # 减掉 self
|
|
|
|
@pytest.mark.parametrize(
|
|
"name",
|
|
[
|
|
"tenant_id",
|
|
"meta",
|
|
"thinking_observation",
|
|
"reasoning_effort",
|
|
"scope",
|
|
"operation",
|
|
"logical_call_id",
|
|
"event_kind",
|
|
"http_status_code",
|
|
"error_type",
|
|
"cause_type",
|
|
"error_body",
|
|
"attempts",
|
|
"total_latency_ms",
|
|
],
|
|
)
|
|
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)
|