diff --git a/pyproject.toml b/pyproject.toml index 35938a9..d746937 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -82,6 +82,7 @@ layers = [ "polygateway.middleware", "polygateway.transports | polygateway.backends | polygateway.telemetry | polygateway.structured", "polygateway.thinking", + "polygateway.deadline", "polygateway.providers : polygateway.sources", "polygateway.ports : polygateway.types : polygateway.errors : polygateway.streaming", ] diff --git a/src/polygateway/__init__.py b/src/polygateway/__init__.py index 52242e0..df2aee2 100644 --- a/src/polygateway/__init__.py +++ b/src/polygateway/__init__.py @@ -10,6 +10,7 @@ from polygateway.config import EmbeddingSettings, GatewaySettings, OcrSettings from polygateway.embedding import EmbeddingClient from polygateway.errors import ( AllSourcesExhausted, + CallDeadlineExceeded, CircuitOpenError, GatewayUnavailableError, GovernanceBackendError, @@ -59,6 +60,7 @@ __all__ = [ "Effort", "AllSourcesExhausted", "CallStats", + "CallDeadlineExceeded", "CircuitOpenError", "EmbeddingClient", "EmbeddingResponse", diff --git a/src/polygateway/client.py b/src/polygateway/client.py index 4534337..fe288fb 100644 --- a/src/polygateway/client.py +++ b/src/polygateway/client.py @@ -20,6 +20,7 @@ from polygateway.backends.memory.breaker import InMemoryGate from polygateway.backends.memory.cache import InMemoryCache from polygateway.backends.memory.limiter import InMemoryLimiter from polygateway.config import GatewaySettings +from polygateway.deadline import ensure_call_deadline, with_call_deadline from polygateway.errors import PolyGatewayError from polygateway.middleware.base import compose from polygateway.middleware.cache import CacheMW @@ -235,10 +236,15 @@ class GatewayClient: structured_strategy: StructuredOutputStrategy | None = None, structured_escalation: StructuredOutputStrategy | None = None, structured_max_retries: int = 1, + call_deadline_s: float | None = None, now: Any = time.monotonic, sleep: Any = asyncio.sleep, rng: Any = random.random, ) -> None: + # 入口即校: 装配错误当场报,不等到第一次调用才炸 + self._call_deadline_s = ensure_call_deadline( + call_deadline_s, "GatewayClient(call_deadline_s=...)" + ) emitter = ( TelemetryEmitter(telemetry, scope=scope, pricing=pricing, text_cap=text_cap) if telemetry is not None @@ -292,6 +298,8 @@ class GatewayClient: self._structured_available = structured_strategy is not None self._terminal = terminal # 内部引用: 装配自省/测试用 self._handler = compose(middlewares, terminal) + # 期限到期需要报出 scope(现之前只传给 RetryMW,未自存) + self._scope = scope # 逻辑调用统计需要同一只注入钟(1.3.5);现之前只传给中间件未自存 self._now = now # 终态行由公开边界统一写出(T3),故边界也需持有 emitter @@ -336,6 +344,7 @@ class GatewayClient: reasoning_effort: Effort | str | None = None, tenant_id: str | None = None, meta: Mapping[str, Any] | None = None, + call_deadline_s: float | None = None, ) -> LLMResponse: """一次治理调用(签名冻结,ARCH §5.2;与三项目 LLMProvider 协议兼容)。 @@ -351,6 +360,11 @@ class GatewayClient: `tenant_id` 与 `meta` 是调用方自定义维度,只进遥测、**不进缓存 key** (租户隔离由 `cache_namespace` 负责,ARCH §7.5);前者享有真实列待遇 (可挂 RLS、可进复合索引),后者是任意 KV 容器(issue #11)。 + + `call_deadline_s` 是本次调用的墙钟硬边界(issue #22): `None` = 继承装配值, + 正数 = 本次覆盖,**不提供"本次关闭"**。它治理的是**等待**: 到期抛 + `CallDeadlineExceeded`,但到期**不等于未产出、未计费**——在途请求可能已发出、 + 已被上游计费,且清理仍在 `finally` 里完成,故返回时刻 = 期限 + 清理耗时。 """ if structured is not None and not self._structured_available: raise ImportError( @@ -377,6 +391,13 @@ class GatewayClient: else coerce_effort(reasoning_effort, origin="chat(reasoning_effort=...)") ) validate_thinking_raw(sampling, effort=effort, wire=None, origin="chat overlay") + # 期限取值与校验必须在创建 awaitable **之前**: 否则非法值抛错时会遗留 + # 未 await 的协程(RuntimeWarning + 资源不释放) + deadline = ( + self._call_deadline_s + if call_deadline_s is None + else ensure_call_deadline(call_deadline_s, "chat(call_deadline_s=...)") + ) # 三项校验均已通过 → 进入统计边界(设计 §3: 输入校验异常在边界之外,保持原行为) context = _CallContext(now=self._now) request = ChatRequest( @@ -395,7 +416,9 @@ class GatewayClient: call_context=context, ) try: - response = await self._handler(request) + response = await with_call_deadline( + self._handler(request), deadline_s=deadline, scope=self._scope + ) except PolyGatewayError as exc: # 统计边界内的一切领域失败均尝试写一条终态行(1.3.5 设计 §6 I3), # 包括已有 attempt 错误行的 RequestRejected / ResultInvalid——两类行描述 @@ -485,6 +508,7 @@ class GatewayClient: structured_strategy=strategy, structured_escalation=escalation, structured_max_retries=settings.structured_max_retries, + call_deadline_s=settings.call_deadline_s, ) _mark_owned_components(client, limiter=limiter, breaker=breaker, telemetry=telemetry) client._owns_cache = cache is None # 缓存后端可以是 None(backend=none),helper 会跳过 diff --git a/src/polygateway/config.py b/src/polygateway/config.py index 27dcc5f..b665103 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -19,6 +19,7 @@ from typing import TYPE_CHECKING from dotenv import dotenv_values from loguru import logger +from polygateway.deadline import ensure_call_deadline from polygateway.types import ( BackpressurePolicy, BreakerConfig, @@ -182,6 +183,11 @@ class GatewaySettings: pricing_path: str | None structured_max_retries: int lease_ttl_s: float + # 一次逻辑调用的**可选**墙钟硬边界(issue #22)。缺省 None = 不启用,行为逐字 + # 等于 1.3.5;有默认值故追加在末尾,不扰动既有位置构造。`EmbeddingSettings.gateway` + # 与 `OcrSettings.gateway` 自动继承。值域由 `_validate_call_deadline` 把关, + # 直接构造、`dataclasses.replace` 与 env 三条路一致 + call_deadline_s: float | None = None def __post_init__(self) -> None: self._normalize() @@ -192,6 +198,7 @@ class GatewaySettings: self._validate_lease() self._validate_stall() self._validate_probe() + self._validate_call_deadline() def _normalize(self) -> None: """把 `from_env` 一直在做的规范化补到构造路上,两条路必须产出同一个值。 @@ -345,6 +352,19 @@ class GatewaySettings: f"timeout_s + {_PROBE_GRACE_S}({floor});调大 probe_ttl_s 或调小源的 timeout_s" ) + def _validate_call_deadline(self) -> None: + """期限值域守卫: 盖住直接构造与 `dataclasses.replace` 两条路(issue #22)。 + + env 路已在 `_load_call_deadline` 里带真实键名报过错, 此处对合法值是幂等空操作。 + **不校验**它与 `timeout_s`/`stall_window_s` 的大小关系: 期限短于单次超时 + 是调用方的合法选择(要的就是“不让这次调用拖过 N 秒”)。 + """ + object.__setattr__( + self, + "call_deadline_s", + ensure_call_deadline(self.call_deadline_s, "GatewaySettings.call_deadline_s"), + ) + @classmethod def from_env( cls, @@ -373,6 +393,7 @@ class GatewaySettings: selector=_load_choice(env, f"{scope_u}__SELECTOR", _SELECTORS, "health_aware"), quota_full=_load_choice(env, f"{scope_u}__QUOTA_FULL", _QUOTA_FULL, "wait"), circuit_open=_load_choice(env, f"{scope_u}__CIRCUIT_OPEN", _CIRCUIT_OPEN, "fail_fast"), + call_deadline_s=_load_call_deadline(scope_u, env), **_load_pgw(env), ) @@ -671,6 +692,30 @@ def _load_lease_ttl(env: Mapping[str, str]) -> float: return float(_cast(found[1], "float", found[0])) if found else _DEFAULT_LEASE_TTL_S +def _load_call_deadline(scope: str, env: Mapping[str, str]) -> float | None: + """读 `{SCOPE}__CALL_DEADLINE_S`(issue #22);键未设即 None = 不启用。 + + 用 `_first` 而非 `_require`: 后者会把"未设"当成配置缺失报错,对存量下游 + 就是破坏性变更。键名两段式(`split("__")` 长度 2 ≠ 4),故 `_load_sources` + 天然跳过它,不必进 `_RESERVED_SEGMENTS`。 + + origin 传**实际命中的 env 键名**而非字段名: `_cast` 只接得住"不是数字", + `0`/负数/`inf` 会穿过它落到 `ensure_call_deadline`——那时报一条指向字段名的错误, + 在多 scope 部署里无法定位是哪个键写错了。 + + Args: + scope: 已大写的 scope 名。 + env: 已合并的环境映射。 + + Returns: + 一次逻辑调用的墙钟期限(秒);键未设或为空串时返回 None(不启用)。 + """ + found = _first(env, f"{scope}__CALL_DEADLINE_S") + if found is None: + return None + return ensure_call_deadline(_cast(found[1], "float", found[0]), found[0]) + + @dataclass(frozen=True) class EmbeddingSettings: """Embedding scope 装配配置(M2 §7): 复用 GatewaySettings + embedding 专用键。 diff --git a/src/polygateway/deadline.py b/src/polygateway/deadline.py new file mode 100644 index 0000000..22e5433 --- /dev/null +++ b/src/polygateway/deadline.py @@ -0,0 +1,80 @@ +"""一次逻辑调用的**可选**墙钟硬边界(issue #22;1.3.6 设计 §3 方案 A)。 + +只依赖标准库与 `errors.py`(依赖铁律最内层),供三个公开边界各包一次: +期限治理的是**等待**,不是"到期即无副作用"——在途请求可能已发出、已被上游 +计费,清理照旧在 `finally` 完成,故返回时刻 = 期限 + 清理耗时。 + +缺省 `None` 时**完全不进上下文管理器**,行为逐字等于 1.3.5。 +""" + +from __future__ import annotations + +import asyncio +import math +from typing import TYPE_CHECKING + +from polygateway.errors import CallDeadlineExceeded + +if TYPE_CHECKING: + from collections.abc import Awaitable + + +def ensure_call_deadline(value: object, origin: str) -> float | None: + """全装配路径共用的期限值域校验: `None` 或**有限正数秒**,否则当场 `ValueError`。 + + 装配错误不属降级面(缺失/非法配置直接报错,不静默取默认值)。`bool` 必须先判: + `isinstance(True, int)` 为真,放行会让 `call_deadline_s=True` 变成"1 秒期限" + 这种没人写得出来的意图。巨大 int(如 `10**400`)超出 float 值域,`float()` 会抛 + `OverflowError`——它不是 `ValueError` 的子类,泄漏出去会绕过调用方的 + `except ValueError`,故在此统一成同一种装配错误。 + + `origin` 写进消息,用于在多 scope 部署里定位到底是哪个键/哪个参数非法。 + """ + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{origin} 必须是 None 或有限正数秒: {value!r}") + try: + seconds = float(value) + except OverflowError: + raise ValueError(f"{origin} 必须是有限正数秒: {value!r}") from None + if not math.isfinite(seconds) or seconds <= 0: + raise ValueError(f"{origin} 必须是有限正数秒: {value!r}") + return seconds + + +async def with_call_deadline[T](aw: Awaitable[T], *, deadline_s: float | None, scope: str) -> T: + """给一个 awaitable 加一层可选期限;到期抛 `CallDeadlineExceeded`。 + + 三条实现红线: + + 1. **校验先于构造 awaitable**——调用方必须先 `ensure_call_deadline`,否则非法值 + 抛错时会遗留未 await 的协程(`RuntimeWarning` + 资源不释放)。 + 2. 只用**相对时长**,绝不把注入的 `now` 换算成绝对截止时刻: 注入钟跳变 + 10^6 秒不该凭空触发期限。 + 3. 判据必须是**局部变量身份比较**,不可退化成只看 `cm.expired()`: + 到期后清理路径自抛的 `TimeoutError` 也发生在 `expired()` 为真时,只看它 + 会把别人的超时改标成本层期限;`__cause__` 启发式同样失效(内层 + `asyncio.timeout` 抛出的 `TimeoutError` 其 `__cause__` 也是 `CancelledError`)。 + + 不新增后台任务、不 `shield`、不改异常对象:外部取消照常以 `CancelledError` 穿透。 + """ + if deadline_s is None: + # 未启用: 不进上下文管理器,逐字走 1.3.5 旧路径 + return await aw + # 体内(含清理路径)自抛的 TimeoutError 的**身份**,唯一可靠的区分依据 + inner_timeout: BaseException | None = None + # 先建对象再进上下文: `as cm` 只在 `__aenter__` 返回后才绑定, 而 except 块无条件 + # 读 `cm`——进入阶段一旦异常就会变成 NameError 掩盖真实错误 + cm = asyncio.timeout(deadline_s) + try: + async with cm: + try: + return await aw + except TimeoutError as exc: + inner_timeout = exc + raise + except TimeoutError as exc: + if cm.expired() and exc is not inner_timeout: + raise CallDeadlineExceeded(scope=scope, deadline_s=deadline_s) from None + raise diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py index e191c2c..a2b0684 100644 --- a/src/polygateway/embedding.py +++ b/src/polygateway/embedding.py @@ -28,6 +28,7 @@ from loguru import logger from polygateway.client import _aclose_component, _telemetry_status_of from polygateway.config import EmbeddingSettings +from polygateway.deadline import ensure_call_deadline, with_call_deadline from polygateway.errors import ( AllSourcesExhausted, GovernanceBackendError, @@ -112,6 +113,7 @@ class EmbeddingClient: batch_size: int, normalize: bool = False, expected_dim: int | None = None, + call_deadline_s: float | None = None, now: Callable[[], float] = time.monotonic, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, rng: Callable[[], float] = random.random, @@ -120,6 +122,10 @@ class EmbeddingClient: raise ValueError("batch_size 必须 ≥ 1") if expected_dim is not None and expected_dim < 1: raise ValueError("expected_dim 必须 ≥ 1") + # 入口即校: 装配错误当场报,不等到第一次调用才炸 + self._call_deadline_s = ensure_call_deadline( + call_deadline_s, "EmbeddingClient(call_deadline_s=...)" + ) self._scope = scope # embed payload 硬编码 {model, input},带 extra_body 的源必须先剥离, # 否则遥测会记录一个从未发出的采样参数(issue #4 决策 G) @@ -174,11 +180,16 @@ class EmbeddingClient: parent_call_id: str | None = None, tenant_id: str | None = None, meta: Mapping[str, Any] | None = None, + call_deadline_s: float | None = None, ) -> EmbeddingResponse: """一次治理 embedding 调用: 按 batch_size 切批,批间串行,全批合并返回。 `tenant_id` 与 `meta` 是调用方自定义维度,只进遥测(issue #11);它们属于 本次调用而非某一批,故每批的遥测行都带同一份维度。 + + `call_deadline_s` 是本次调用的墙钟硬边界(issue #22): `None` = 继承装配值。 + **整次调用共享一份**——N 批串行跑在同一条期限内,不按批数放大 N 倍。 + `texts == []` 的早返回在期限之外(零尝试,无等待可治)。 """ if not isinstance(texts, list) or any(not isinstance(t, str) for t in texts): raise TypeError("texts 必须是 list[str](显式优于隐式,不收单条 str)") @@ -188,6 +199,12 @@ class EmbeddingClient: dimension_tenant_id, dimensions = validate_caller_dimensions( tenant_id, meta, origin="embed(tenant_id=..., meta=...)" ) + # 期限取值与校验必须在创建 awaitable **之前**(否则遗留未 await 的协程) + deadline = ( + self._call_deadline_s + if call_deadline_s is None + else ensure_call_deadline(call_deadline_s, "embed(call_deadline_s=...)") + ) # 校验均已通过 → 进入统计边界(设计 §3.5: `texts` 类型与调用方维度校验之后) context = _CallContext(now=self._now) if not texts: @@ -206,8 +223,12 @@ class EmbeddingClient: call_stats=context.snapshot(), ) try: - return await self._embed_all( - texts, session_id, parent_call_id, dimension_tenant_id, dimensions, context + return await with_call_deadline( + self._embed_all( + texts, session_id, parent_call_id, dimension_tenant_id, dimensions, context + ), + deadline_s=deadline, + scope=self._scope, ) except PolyGatewayError as exc: await self._emit_terminal( @@ -634,6 +655,7 @@ class EmbeddingClient: batch_size=settings.batch_size, normalize=settings.normalize, expected_dim=settings.expected_dim, + call_deadline_s=gw.call_deadline_s, ) _mark_owned_components(client, limiter=limiter, breaker=breaker, telemetry=telemetry) return client diff --git a/src/polygateway/errors.py b/src/polygateway/errors.py index 77d6033..1fd324b 100644 --- a/src/polygateway/errors.py +++ b/src/polygateway/errors.py @@ -217,3 +217,23 @@ class GovernanceBackendError(GatewayUnavailableError): # 父类会把 message 覆写为 "{scope} 网关暂时不可用: {reason}",而各构造点 # 携带的诊断串(如"限流后端 try_acquire 失败: ...")是排障主线索,必须保住 self.args = (message,) + + +class CallDeadlineExceeded(PolyGatewayError): # noqa: N818 — 设计 §9 人类批准的公共名 + """调用方设定的整体调用期限到期; 不是网关不可用、也不是源故障。 + + 刻意**不属**四分类、**不进** `SCOPE_REASONS`、**不继承** `GatewayUnavailableError`: + 它描述的是调用方自己的耐心边界, 与"对方怎么了"正交——按四分类之一上报会让 + 下游的重试/换源/熔断逻辑对着一次本地超时做治理决策(库铁律「错误分类驱动」)。 + + 也刻意**没有** `retry_after_s`: 期限到期不含"何时可再试"的信息, 给 `0.0` + 会按既定语义指示下游立刻重打一条可能已经饱和的通道。 + + **到期不等于未产出、未计费**: 期限治理的是等待, 在途请求可能已经发出、 + 已被上游计费, 清理仍在 `finally` 里完成, 故返回时刻 = 期限 + 清理耗时。 + """ + + def __init__(self, *, scope: str, deadline_s: float) -> None: + super().__init__(f"{scope} 调用期限 {deadline_s}s 到期") + self.scope = scope + self.deadline_s = deadline_s diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index 2c282d7..4c6bbca 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -23,6 +23,7 @@ from typing import TYPE_CHECKING, Any, Literal from loguru import logger from polygateway.client import _aclose_component, _telemetry_status_of +from polygateway.deadline import ensure_call_deadline, with_call_deadline from polygateway.errors import ( AllSourcesExhausted, GovernanceBackendError, @@ -115,10 +116,15 @@ class OcrClient: circuit_open: str = "fail_fast", telemetry: TelemetryRecorder | None = None, text_cap: int | None = None, + call_deadline_s: float | None = None, now: Callable[[], float] = time.monotonic, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, rng: Callable[[], float] = random.random, ) -> None: + # 入口即校: 装配错误当场报,不等到第一次调用才炸 + self._call_deadline_s = ensure_call_deadline( + call_deadline_s, "OcrClient(call_deadline_s=...)" + ) self._scope = scope # MonkeyOCR 只发 multipart 表单,带 extra_body 的源必须先剥离,否则 # 遥测会记录一个从未发出的采样参数(issue #4 决策 G) @@ -171,10 +177,14 @@ class OcrClient: parent_call_id: str | None = None, tenant_id: str | None = None, meta: Mapping[str, Any] | None = None, + call_deadline_s: float | None = None, ) -> OcrTextResult: """一次治理文本转录(/ocr/text);text 空串 = 合法"无文字"。 `tenant_id` 与 `meta` 是调用方自定义维度,只进遥测(issue #11)。 + + `call_deadline_s` 是本次调用的墙钟硬边界(issue #22): `None` = 继承装配值。 + 它治理的是**等待**——到期不等于未产出,返回时刻 = 期限 + 清理耗时。 """ # 必须在进链路之前校验: 链路内的一切失败都被遥测层降级成 warning # (库铁律「遥测写失败降级不冒泡」),校验放下游等于没有校验 @@ -189,6 +199,7 @@ class OcrClient: parent_call_id, dimension_tenant_id, dimensions, + call_deadline_s, ) result = outcome.result return OcrTextResult( @@ -209,10 +220,13 @@ class OcrClient: parent_call_id: str | None = None, tenant_id: str | None = None, meta: Mapping[str, Any] | None = None, + call_deadline_s: float | None = None, ) -> OcrLayoutResult: """一次治理版面解析(/parse → ZIP);elements 空 = 合法"无元素"。 `tenant_id` 与 `meta` 是调用方自定义维度,只进遥测(issue #11)。 + + `call_deadline_s` 同 `recognize_text`(issue #22): `None` = 继承装配值。 """ # 校验早于链路,理由同 recognize_text;origin 标明方法名以便定位入口 dimension_tenant_id, dimensions = validate_caller_dimensions( @@ -226,6 +240,7 @@ class OcrClient: parent_call_id, dimension_tenant_id, dimensions, + call_deadline_s, ) result = outcome.result return OcrLayoutResult( @@ -262,17 +277,28 @@ class OcrClient: parent_call_id: str | None, tenant_id: str | None, meta: dict[str, Any], + call_deadline_s: float | None = None, ) -> tuple[_AttemptOutcome, CallStats]: if not isinstance(image, bytes): raise TypeError("image 必须是 bytes(路径读取/批量拼帧留业务侧,D9)") if not image: raise ValueError("image 不能为空") + # 期限校验与 `image` 校验同列(仍在 `_CallContext` 之前、创建 awaitable 之前) + deadline = ( + self._call_deadline_s + if call_deadline_s is None + else ensure_call_deadline(call_deadline_s, f"{operation}(call_deadline_s=...)") + ) # M1 例外: `image` 校验在 `_call` 内而非公开方法,故上下文在该校验 # **通过之后**创建——这样设计 §3 的"校验在统计边界外"对 OCR 才成立 context = _CallContext(now=self._now) try: - return await self._run( - kind, operation, image, session_id, parent_call_id, tenant_id, meta, context + return await with_call_deadline( + self._run( + kind, operation, image, session_id, parent_call_id, tenant_id, meta, context + ), + deadline_s=deadline, + scope=self._scope, ) except PolyGatewayError as exc: await self._emit_terminal( @@ -670,6 +696,7 @@ class OcrClient: # OCR 行与 chat 行写同一张 llm_calls;漏传这一条,同表内就一半受控 # 一半不受控(issue #12) text_cap=gw.telemetry_text_cap, + call_deadline_s=gw.call_deadline_s, ) _mark_owned_components(client, limiter=limiter, breaker=breaker, telemetry=telemetry) return client diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index c9451ce..5d8cc80 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -1,8 +1,10 @@ """GatewayClient 装配与端到端(fake 后端 + MockTransport)测试(设计 §2.4)。""" import asyncio +import gc import json import sys +import warnings from pathlib import Path import httpx @@ -10,10 +12,12 @@ import pytest from polygateway import ( AllSourcesExhausted, + CallDeadlineExceeded, GatewayClient, GatewaySettings, RequestRejectedError, ResultInvalidError, + TransientError, gather_bounded, ) from polygateway.backends.memory.breaker import InMemoryGate @@ -33,6 +37,10 @@ from polygateway.types import ( SourceConfig, ) +# 复用 RetryMW 那份可编程 fake transport(含确定性 `entered` 窗口),不再造第二份; +# `tests/unit/test_backpressure.py:34` 已是同款复用 +from tests.unit.test_retry import FakeTransport, _ok + _REPO = Path(__file__).resolve().parents[2] _ENV = { @@ -1717,3 +1725,207 @@ class TestTerminalEmitDegradation: error=AllSourcesExhausted(scope="llm", reason="retry_exhausted", retry_after_s=1.0), operation="chat", ) + + +class _ClockJumpTransport: + """假 transport: 只推进**注入钟**,真实墙钟几乎不走。 + + 用于把"期限读哪只钟"与"统计读哪只钟"两件事分开断言。 + """ + + def __init__(self, clock, *, jump): + self._clock = clock + self._jump = jump + self.calls = [] + + async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort): + self.calls.append(call_id) + self._clock.advance(self._jump) + return _ok() + + +class _SlowRecorder: + """假 recorder: 每写一行真实等待一段,用于量化"清理不被期限截断"。""" + + def __init__(self, delay=0.15): + self._delay = delay + self.rows = [] + + async def record_llm_call(self, **fields): + await asyncio.sleep(self._delay) + self.rows.append(fields) + + +class _SlowSetCache: + """假缓存后端: `set` 慢于期限,用于构造"已产出、已计费的成功被丢弃"。""" + + def __init__(self, delay=0.5): + self._delay = delay + self.data = {} + self.sets = 0 + + async def get(self, key): + return self.data.get(key) + + async def set(self, key, value, ttl_s): + self.sets += 1 + await asyncio.sleep(self._delay) + self.data[key] = value + + +class TestChatCallDeadline: + """chat 链路的期限覆盖面与到期代价(计划 §5 批次 D/D2)。 + + 真实事件循环时钟: 期限 0.05s 对被治理的等待(退避 5s、轮询 300s、慢 IO 0.5s) + 有 10 倍以上余量,故不标 slow。 + """ + + _MSG = [{"role": "user", "content": "hi"}] + _DEADLINE = 0.05 + + def _rows(self, recorder, kind): + return [r for r in recorder.rows if r["event_kind"] == kind] + + # —— 批次 D: 期限落点覆盖面 —— + + async def test_deadline_fires_during_backoff_sleep(self): + """退避 sleep 是等待的大头(429 序列可睡到小时级),期限必须能在它中间落地。""" + transport = FakeTransport([TransientError("boom", operation="chat"), _ok()]) + async with _client(transport=transport, retry=RetryPolicy(3, 5.0, 30.0)) as client: + with pytest.raises(CallDeadlineExceeded) as exc: + await client.chat(self._MSG, call_deadline_s=self._DEADLINE) + assert exc.value.scope == "llm" and exc.value.deadline_s == self._DEADLINE + # 第二次尝试还压在 5s 退避里,期限确实落在 sleep 上而非 transport 上 + assert len(transport.calls) == 1 + + async def test_deadline_fires_while_queued_for_quota(self): + """准入排队(配额满轮询)是第二类长等待: 一次 transport 都没打出去也要能到期。""" + source = _source(tpm=1, est_tokens=1000) # 预扣量恒超本源 TPM → 六闸永不放行 + limiter = InMemoryLimiter( + scope="llm", sources={source.name: source}, global_limits=GlobalLimits(0, 0, 0) + ) + transport = FakeTransport([_ok()]) + async with _client([source], transport=transport, limiter=limiter) as client: + with pytest.raises(CallDeadlineExceeded): + await client.chat(self._MSG, call_deadline_s=self._DEADLINE) + assert transport.calls == [] # 期限落在轮询里,尝试从未开始 + + async def test_deadline_fires_during_structured_re_ask(self): + """结构化重问共享同一份期限: 阶梯不得按轮数各起一份,否则期限被放大 N 倍。 + + 窗口用"第三轮挂起"构造而非 sleep 猜时长——前两轮瞬时返回坏 JSON, + 期限只可能落在第三轮上,断言因此与机器负载无关。 + """ + from pydantic import BaseModel + + class Answer(BaseModel): + answer: int + + transport = FakeTransport([_ok("not json at all"), _ok("not json at all"), "hang"]) + async with _client( + transport=transport, structured_max_retries=5, structured_strategy=JsonRepairStrategy() + ) as client: + with pytest.raises(CallDeadlineExceeded): + await client.chat(self._MSG, structured=Answer, call_deadline_s=0.05) + # 到期发生在第三轮: 期限确实跨过了两次重问,而不是在首轮就截断 + assert len(transport.calls) == 3 + + # —— 批次 D2: 到期代价 —— + + async def test_expiry_writes_one_terminal_row_and_a_cancelled_attempt(self): + """到期恰好一条终态行 + 被取消的 attempt 行,两行同一 logical_call_id。""" + recorder = _MemoryRecorder() + transport = FakeTransport(["hang"]) + async with _client(transport=transport, telemetry=recorder) as client: + with pytest.raises(CallDeadlineExceeded): + await client.chat(self._MSG, call_deadline_s=self._DEADLINE) + attempts = self._rows(recorder, "attempt") + terminals = self._rows(recorder, "terminal_failure") + assert len(terminals) == 1 + assert terminals[0]["error_type"] == "CallDeadlineExceeded" + assert len(attempts) == 1 and attempts[0]["error"] == "cancelled" + assert attempts[0]["logical_call_id"] == terminals[0]["logical_call_id"] + + # 零新增遥测列: 期限终态行的列集合与既有失败路径的终态行逐字相同 + baseline = _MemoryRecorder() + async with _client( + handler=lambda request: httpx.Response(400, json={"error": {"message": "bad"}}), + telemetry=baseline, + ) as client: + with pytest.raises(RequestRejectedError): + await client.chat(self._MSG) + assert set(terminals[0]) == set(self._rows(baseline, "terminal_failure")[0]) + + async def test_cleanup_is_not_cut_short_by_the_expiry(self): + """返回时刻 = 期限 + 清理耗时: 只断下界(> 期限 × 2),不断上界。""" + recorder = _SlowRecorder(delay=0.15) # attempt 行与终态行各付一次 + transport = FakeTransport(["hang"]) + loop = asyncio.get_running_loop() + started = loop.time() + async with _client(transport=transport, telemetry=recorder) as client: + with pytest.raises(CallDeadlineExceeded): + await client.chat(self._MSG, call_deadline_s=self._DEADLINE) + elapsed = loop.time() - started + assert len(recorder.rows) == 2 # 清理照常写完两行,没被期限截断 + assert elapsed > self._DEADLINE * 2, f"清理疑似被截断: {elapsed}s" + + async def test_expiry_discards_a_success_that_was_already_billed(self): + """到期 ≠ 未产出、未计费: transport 已成功一次,结果仍被丢弃。""" + transport = FakeTransport([_ok()]) + cache = _SlowSetCache(delay=0.5) # 写缓存慢于期限 → 到期落在成功之后 + async with _client( + transport=transport, cache=cache, cache_namespace="proj", cache_ttl_s=600 + ) as client: + with pytest.raises(CallDeadlineExceeded): + await client.chat(self._MSG, call_deadline_s=self._DEADLINE) + assert len(transport.calls) == 1 # 上游已经产出并计费 + assert cache.sets == 1 and cache.data == {} # 结果既没回给调用方也没落缓存 + + # —— 批次 E: 注入钟与期限正交 —— + + async def test_injected_clock_jump_does_not_trigger_the_deadline(self): + """期限只认真实墙钟: 注入钟跳 10^6 秒也不该凭空到期(不换算绝对截止时刻)。""" + clock = _StatsClock() + transport = _ClockJumpTransport(clock, jump=1_000_000.0) + async with _client(transport=transport, now=clock) as client: + resp = await client.chat(self._MSG, call_deadline_s=5.0) + assert resp.content == "ok" + # 而统计仍逐字读注入钟(10^6 s = 10^9 ms),两只钟各司其职 + assert resp.call_stats is not None + assert resp.call_stats.total_latency_ms == 1_000_000_000 + + async def test_expiry_latency_still_reads_the_injected_clock(self): + """期限由真实钟触发,终态行的耗时仍取自注入钟(真实耗时只有几十毫秒)。""" + clock = _StatsClock() + recorder = _TickingRecorder(clock, tick=0.5) + transport = FakeTransport(["hang"]) + async with _client(transport=transport, telemetry=recorder, now=clock) as client: + with pytest.raises(CallDeadlineExceeded): + await client.chat(self._MSG, call_deadline_s=self._DEADLINE) + terminal = [r for r in recorder.rows if r["event_kind"] == "terminal_failure"][0] + assert terminal["total_latency_ms"] == 500 # attempt 行那一次 tick,不是真实的 ~50ms + + +class TestChatCallDeadlineEntryGuards: + """per-call 入口校验的两条硬红线(计划 §3.4/§5 批次 E)。""" + + _MSG = [{"role": "user", "content": "hi"}] + + async def test_illegal_per_call_value_leaves_no_un_awaited_coroutine(self): + """校验先于构造 awaitable: 否则非法值抛错时遗留未 await 的协程(资源不释放)。""" + transport = FakeTransport([_ok()]) + async with _client(transport=transport) as client: + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + with pytest.raises(ValueError, match=r"chat\(call_deadline_s"): + await client.chat(self._MSG, call_deadline_s=0) + gc.collect() # 未 await 的协程在回收时才发 RuntimeWarning + assert [w for w in caught if "never awaited" in str(w.message)] == [] + assert transport.calls == [] + + async def test_per_call_none_inherits_the_assembled_deadline(self): + """`None` = 继承装配值(不提供"本次关闭"): 装配了期限就照样到期。""" + transport = FakeTransport(["hang"]) + async with _client(transport=transport, call_deadline_s=0.05) as client: + with pytest.raises(CallDeadlineExceeded): + await client.chat(self._MSG) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index e853783..5a7ec82 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -1041,3 +1041,58 @@ def test_live_unknown_wire_assembly_is_local_only(): GatewayClient.from_settings( dataclasses.replace(settings, sources=(source,)), registry=register_provider(mystery) ) + + +class TestCallDeadlineConfig: + """`{SCOPE}__CALL_DEADLINE_S` 与三个 client 入口参数的值域四条路(issue #22)。""" + + def test_key_unset_means_disabled(self): + assert GatewaySettings.from_env("LLM", env=_env()).call_deadline_s is None + + def test_env_key_parsed(self): + s = GatewaySettings.from_env("LLM", env=_env(**{"LLM__CALL_DEADLINE_S": "30"})) + assert s.call_deadline_s == 30.0 + + @pytest.mark.parametrize("bad", ["0", "-1", "nan", "inf", "abc"]) + def test_env_illegal_value_reports_the_actual_key_name(self, bad): + """origin 必须是实际命中的 env 键名,多 scope 部署里才定位得到是哪个键。""" + with pytest.raises(ValueError, match="LLM__CALL_DEADLINE_S"): + GatewaySettings.from_env("LLM", env=_env(**{"LLM__CALL_DEADLINE_S": bad})) + + def test_direct_construction_is_guarded(self): + base = GatewaySettings.from_env("LLM", env=_env()) + with pytest.raises(ValueError, match="GatewaySettings.call_deadline_s"): + dataclasses.replace(base, call_deadline_s=0) + + def test_plain_constructor_call_is_guarded_too(self): + """`dataclasses.replace` 与直接构造是两条路: 守卫在 `__post_init__` 才两条都盖住。 + + 只在 `from_env` 里校验的话,直接 `GatewaySettings(...)` 装配的下游(测试/高级 + 注入路径,CLAUDE.md §4.5 的第二条装配路)会把非法期限一路带到第一次调用才炸。 + """ + base = GatewaySettings.from_env("LLM", env=_env()) + fields = {f.name: getattr(base, f.name) for f in dataclasses.fields(base)} + with pytest.raises(ValueError, match="GatewaySettings.call_deadline_s"): + GatewaySettings(**{**fields, "call_deadline_s": float("inf")}) + # 合法值走同一条路不受影响(守卫对合法值是幂等空操作) + assert GatewaySettings(**{**fields, "call_deadline_s": 7}).call_deadline_s == 7.0 + + def test_replace_with_legal_value_is_idempotent(self): + base = GatewaySettings.from_env("LLM", env=_env()) + assert dataclasses.replace(base, call_deadline_s=5).call_deadline_s == 5.0 + + def test_deadline_shorter_than_timeout_is_legal(self): + """期限短于单次 timeout_s 是调用方的合法选择,不做跨字段耦合校验。""" + s = GatewaySettings.from_env("LLM", env=_env(**{"LLM__CALL_DEADLINE_S": "1"})) + assert s.call_deadline_s == 1.0 and s.sources[0].timeout_s == 120.0 + + def test_from_settings_propagates_to_client(self): + s = GatewaySettings.from_env("LLM", env=_env(**{"LLM__CALL_DEADLINE_S": "12"})) + assert GatewayClient.from_settings(s)._call_deadline_s == 12.0 + + def test_client_init_validates_at_entry(self): + """三个 client 的 `__init__` 直传非法值也当场报错(不经 settings 那道守卫)。""" + from tests.unit.test_client import _client + + with pytest.raises(ValueError, match=r"GatewayClient\(call_deadline_s"): + _client(call_deadline_s=0) diff --git a/tests/unit/test_deadline.py b/tests/unit/test_deadline.py new file mode 100644 index 0000000..cce6abc --- /dev/null +++ b/tests/unit/test_deadline.py @@ -0,0 +1,142 @@ +"""`deadline.py` 值域校验与五种形态区分测试(计划 §5 批次 A/B)。 + +用真实事件循环时钟(期限 0.05s、体 0.3s,4-10 倍余量),不标 slow: +被测对象是"哪一种 TimeoutError"的身份判据,注入钟无法覆盖 `asyncio.timeout`。 +""" + +import asyncio + +import pytest + +from polygateway.deadline import ensure_call_deadline, with_call_deadline +from polygateway.errors import CallDeadlineExceeded + +# —— 批次 A: 值域 —— + + +def test_ensure_call_deadline_accepts_none_and_positive(): + assert ensure_call_deadline(None, "origin") is None + assert ensure_call_deadline(3, "origin") == 3.0 + assert ensure_call_deadline(0.5, "origin") == 0.5 + + +@pytest.mark.parametrize( + "bad", + [0, 0.0, -1, -0.5, float("nan"), float("inf"), float("-inf"), "1", True, False, object(), []], +) +def test_ensure_call_deadline_rejects_out_of_range(bad): + with pytest.raises(ValueError) as exc: + ensure_call_deadline(bad, "GatewayClient(call_deadline_s=...)") + assert "GatewayClient(call_deadline_s=...)" in str(exc.value) + + +def test_ensure_call_deadline_rejects_huge_int_without_leaking_overflow(): + """超出 float 值域的巨大 int 也统一 ValueError,不泄漏 OverflowError。""" + with pytest.raises(ValueError) as exc: + ensure_call_deadline(10**400, "origin") + assert "origin" in str(exc.value) + + +# —— 批次 B: 五种形态 —— + + +async def test_deadline_expiry_raises_call_deadline_exceeded(): + async def body(): + await asyncio.sleep(0.3) + + with pytest.raises(CallDeadlineExceeded) as exc: + await with_call_deadline(body(), deadline_s=0.05, scope="llm") + assert exc.value.scope == "llm" + assert exc.value.deadline_s == 0.05 + + +async def test_inner_timeout_before_expiry_propagates_as_is(): + async def body(): + async with asyncio.timeout(0.01): + await asyncio.sleep(0.3) + + with pytest.raises(TimeoutError) as exc: + await with_call_deadline(body(), deadline_s=5.0, scope="llm") + assert not isinstance(exc.value, CallDeadlineExceeded) + + +async def test_cleanup_timeout_after_expiry_is_not_relabelled(): + """到期后清理路径自抛 TimeoutError → 原样上抛(钉住身份比较,不看 expired())。""" + + async def body(): + try: + await asyncio.sleep(0.3) + except asyncio.CancelledError: + raise TimeoutError("cleanup") from None + + with pytest.raises(TimeoutError) as exc: + await with_call_deadline(body(), deadline_s=0.05, scope="llm") + assert not isinstance(exc.value, CallDeadlineExceeded) + assert str(exc.value) == "cleanup" + + +async def test_external_cancel_before_expiry_propagates_cancelled(): + entered = asyncio.Event() + + async def body(): + entered.set() + await asyncio.sleep(0.3) + + task = asyncio.create_task(with_call_deadline(body(), deadline_s=5.0, scope="llm")) + await entered.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +async def test_external_cancel_after_expiry_propagates_cancelled(): + """到期已在途、外部又取消 → 仍是 CancelledError(取消优先,不被改标)。""" + + started = asyncio.Event() + + async def body(): + started.set() + try: + await asyncio.sleep(0.3) + except asyncio.CancelledError: + await asyncio.sleep(0.2) # 清理期,期间遭外部取消 + raise + + task = asyncio.create_task(with_call_deadline(body(), deadline_s=0.05, scope="llm")) + await started.wait() + await asyncio.sleep(0.1) # 让期限先到期,进入清理 + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +async def test_narrow_success_returns_value_without_pending_cancellation(): + async def body(): + await asyncio.sleep(0.01) + return "ok" + + async def runner(): + return await with_call_deadline(body(), deadline_s=0.2, scope="llm") + + task = asyncio.create_task(runner()) + assert await task == "ok" + assert task.cancelling() == 0 + + +async def test_domain_error_inside_window_propagates(): + class BoomError(RuntimeError): + pass + + async def body(): + raise BoomError("boom") + + with pytest.raises(BoomError): + await with_call_deadline(body(), deadline_s=0.05, scope="llm") + + +async def test_none_deadline_takes_the_legacy_path(): + async def body(): + await asyncio.sleep(0.05) + return "ok" + + assert await with_call_deadline(body(), deadline_s=None, scope="llm") == "ok" diff --git a/tests/unit/test_embedding.py b/tests/unit/test_embedding.py index 624d1d6..0a2c825 100644 --- a/tests/unit/test_embedding.py +++ b/tests/unit/test_embedding.py @@ -13,6 +13,7 @@ import pytest from loguru import logger from polygateway.errors import ( + CallDeadlineExceeded, RequestRejectedError, ResultInvalidError, SourceDeadError, @@ -662,3 +663,70 @@ class TestEmbedLogicalCallStats: assert resp.call_stats.attempts == 0 assert resp.call_stats.logical_call_id # 真实 ID,不是空串 assert rec.rows == [] # 零遥测行 + + +class _SlowEmbedTransport: + """假 embedding transport: 每批真实耗时 `delay` 秒。 + + "N 批共享一份期限"只能用真实等待来证——`asyncio.timeout` 认的是事件循环 + 时钟,注入钟推不动它(计划 §5 批次 D)。 + """ + + def __init__(self, *, delay): + self._delay = delay + self.calls = [] + + async def embed(self, *, texts, source, call_id): + self.calls.append(list(texts)) + await asyncio.sleep(self._delay) + return _vec_for(texts) + + +class TestEmbedCallDeadline: + """embedding 的期限语义: 整次调用一份,空输入豁免(计划 §5 批次 D/E)。""" + + async def test_one_deadline_is_shared_across_all_batches(self): + """按批各起一份会让期限被批数放大 N 倍: 单批 0.05s 远小于期限 0.5s 时将永不到期。 + + 余量刷到 10 倍(单批 0.05s vs 期限 0.5s): 要报假结论得单批慢 10 倍, + 而不是机器抳一下就变色。 + """ + transport = _SlowEmbedTransport(delay=0.05) + client, _ = _embed_client([_src()], [], batch_size=1, transport=transport) + loop = asyncio.get_running_loop() + started = loop.time() + with pytest.raises(CallDeadlineExceeded) as exc: + await client.embed([str(i) for i in range(20)], call_deadline_s=0.5) + elapsed = loop.time() - started + assert exc.value.scope == "embed" + # 按批计的话 20 批全都能跑完(根本不会抛),共享一份则跑不到头 + assert 2 <= len(transport.calls) < 20 + assert elapsed < 20 * 0.05, f"总时长疑似随批数放大: {elapsed}s" + + async def test_empty_input_is_exempt_from_the_deadline(self): + """`texts == []` 早返回在 try 之外(零尝试、无等待可治),再小的期限也不该拦它。""" + transport = ScriptedEmbedTransport([]) + client, _ = _embed_client([_src()], [], transport=transport) + resp = await client.embed([], call_deadline_s=1e-6) + assert resp.vectors == [] + assert resp.call_stats is not None and resp.call_stats.attempts == 0 + assert transport.calls == [] + + async def test_the_same_tiny_deadline_does_fire_on_a_non_empty_input(self): + """对照组: 上一条用的 1e-6 秒确实是会到期的值,豁免不是因为期限没生效。""" + transport = _SlowEmbedTransport(delay=0.05) + client, _ = _embed_client([_src()], [], transport=transport) + with pytest.raises(CallDeadlineExceeded): + await client.embed(["a"], call_deadline_s=1e-6) + + async def test_illegal_per_call_value_is_rejected_at_the_entry(self): + """per-call 非法值当场 ValueError,且消息指向 `embed(...)` 而非某个 env 键。""" + transport = ScriptedEmbedTransport([]) + client, _ = _embed_client([_src()], [], transport=transport) + with pytest.raises(ValueError, match=r"embed\(call_deadline_s"): + await client.embed(["a"], call_deadline_s=0) + assert transport.calls == [] + + def test_illegal_constructor_value_is_rejected_at_assembly(self): + with pytest.raises(ValueError, match=r"EmbeddingClient\(call_deadline_s"): + _embed_client([_src()], [], call_deadline_s=-1) diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index 438b0f9..a72caae 100644 --- a/tests/unit/test_ocr_client.py +++ b/tests/unit/test_ocr_client.py @@ -13,6 +13,7 @@ from polygateway.backends.memory.breaker import InMemoryGate from polygateway.backends.memory.limiter import InMemoryLimiter from polygateway.errors import ( AllSourcesExhausted, + CallDeadlineExceeded, CircuitOpenError, RequestRejectedError, ResultInvalidError, @@ -669,3 +670,29 @@ class TestOcrLogicalCallStats: await client.recognize_text("not-bytes") with pytest.raises(ValueError): await client.recognize_text(b"") + + +class TestOcrCallDeadline: + """OCR 两个公开入口的期限与 per-call 校验(计划 §3.4/§5 批次 D/E)。""" + + async def test_expiry_on_a_hanging_transport(self): + client, limiter, _ = _client([_src()], ["hang"]) + with pytest.raises(CallDeadlineExceeded) as exc: + await client.recognize_text(b"jpg", call_deadline_s=0.05) + assert exc.value.scope == "ocr" and exc.value.deadline_s == 0.05 + # 清理照常在 finally 完成: 在途计数必须归零(OCR 无 token,结算恒 0) + stats = await limiter.source_stats("m1") + assert stats.inflight == 0 and stats.tpm_used == 0 + + @pytest.mark.parametrize("method", ["recognize_text", "parse_layout"]) + async def test_illegal_per_call_value_names_the_entry_it_came_from(self, method): + """两个入口各自报自己的名字: 多入口部署里才定位得到是哪次调用传错了。""" + transport = ScriptedOcrTransport([]) + client, _, _ = _client([_src()], [], transport=transport) + with pytest.raises(ValueError, match=rf"{method}\(call_deadline_s"): + await getattr(client, method)(b"jpg", call_deadline_s=float("inf")) + assert transport.calls == [] + + def test_illegal_constructor_value_is_rejected_at_assembly(self): + with pytest.raises(ValueError, match=r"OcrClient\(call_deadline_s"): + _client([_src()], [], call_deadline_s=0)