From 901a19351483ea3533691bee15cf071f9428c9a2 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 14 Sep 2026 00:35:35 -0400 Subject: [PATCH] feat: add a content event and stall accounting to the transport port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T1(计划 §4):观测原语与端口契约(issue #27/#28 底层)。 - Transport.complete 加必填 kw content_event(顺序在 first_token_event 后,无默认值,同款约定):流式每增量 chunk 置位一次,非流式/对冲路 恒不置位,None = 不观测 - TransportResult 增 stall_ms/stall_count(带默认 None):超阈值相邻 增量 chunk 沉默总额与段数,chunk<2/非流式 → None,尾部不计 - SourceConfig 增 stall_threshold_ms(默认 1000,正整数拒 bool, _validate_watchdog 同款落点) - _complete_stream 记增量 chunk 时间戳(空 chunk/usage 帧/keepalive 不产生记录点)+ 模块级纯函数 _stall_from_chunk_times(严格大于 阈值才减);max_inter_token_ms 原语义一字不动 - §2.2 同步清单全量:11 实现 + 11 调用点(重跑 grep 复核零遗漏) - 批次 A 12 用例先红(TypeError/ImportError/AttributeError)后绿, 证据 tests/outputs/138/t1/ --- src/polygateway/middleware/retry.py | 9 +- src/polygateway/ports.py | 7 + src/polygateway/transports/openai_compat.py | 48 ++++- src/polygateway/types.py | 31 +++ tests/e2e/conftest.py | 2 + .../test_redis_cross_connection.py | 11 +- tests/unit/test_backpressure.py | 11 +- tests/unit/test_client.py | 22 +- tests/unit/test_hedge.py | 14 +- tests/unit/test_live_evidence.py | 2 + tests/unit/test_openai_compat.py | 200 +++++++++++++++++- tests/unit/test_ports.py | 11 +- tests/unit/test_retry.py | 23 +- tests/unit/test_types.py | 19 ++ tests/unit/test_usage_source_domain.py | 1 + 15 files changed, 398 insertions(+), 13 deletions(-) diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 6540679..0350d37 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -273,6 +273,7 @@ class RetryMW: reasons, attempt_fails, first_token_event=None, + content_event=None, generation_sink=generation_sink, ) else: @@ -312,6 +313,7 @@ class RetryMW: attempt_fails: dict[str, int], *, first_token_event: asyncio.Event | None, + content_event: asyncio.Event | None, generation_sink: list[int], ) -> LLMResponse | _Failed: call_id = str(uuid.uuid4()) @@ -342,6 +344,9 @@ class RetryMW: # 对冲编排(计划 §3.4): 原路携带首 token 事件,对冲路恒 None(v1 单路, # 不再梯次);未启用对冲时 __call__ 传 None = 调用方不观测首 token first_token_event=first_token_event, + # 中段沉默处置位(1.3.8): 本阶段(T1)所有调用点恒传 None = 不观测; + # T4 才在启用 stall 对冲的流式原路上接真事件(设计 §5.2 N3③) + content_event=content_event, ) generation_sink.append(int((self._now() - gen_started) * 1000)) if result.usage_source == "unavailable": @@ -434,6 +439,7 @@ class RetryMW: reasons, attempt_fails, first_token_event=first_token, + content_event=None, generation_sink=sink_p, ) ) @@ -473,7 +479,7 @@ class RetryMW: self._record_generation(request, sink_p) return outcome # Phase 4 启动对冲路(v1 单路,H5): 对冲路恒传 first_token_event=None, - # 不再触发梯次对冲 + # 不再触发梯次对冲;content_event 同恒 None(N3③,v1 不梯次) sink_h: list[int] = [] hedge = asyncio.create_task( self._attempt( @@ -482,6 +488,7 @@ class RetryMW: reasons, attempt_fails, first_token_event=None, + content_event=None, generation_sink=sink_h, ) ) diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index 7b595a1..6617509 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -51,6 +51,12 @@ class Transport(Protocol): `first_token_event` 同一约定(1.3.7 对冲 H2): `None` = 调用方不观测首 token (未启用对冲);非流式实现**永不置位**(物理上无中途信号,事件自然退化为纯时间 阈值),流式实现在首个增量(内容或思考)到达时置位。 + + `content_event` 同一约定(1.3.8 中段沉默对冲,issue #27/#28): `None` = 调用方 + 不观测增量 chunk(未启用中段对冲);**流式**实现每到达一个增量 chunk(`content` + 或 `reasoning_content`,与看门狗/首 token 活性口径一致)置位一次;空 chunk、 + usage 帧、SSE keepalive 不算;非流式实现**永不置位**;对冲路恒收 `None` + (v1 不梯次,设计 §5.2 N3③)。 """ async def complete( @@ -63,6 +69,7 @@ class Transport(Protocol): call_id: str, reasoning_effort: Effort | None, first_token_event: asyncio.Event | None, + content_event: asyncio.Event | None, ) -> TransportResult: ... diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index 554141b..23e2f47 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -85,6 +85,29 @@ def _sse_delta(chunk: dict[str, Any], usage_sink: dict[str, Any]) -> tuple[bool, return None +def _stall_from_chunk_times( + times: list[float], threshold_ms: int +) -> tuple[float | None, float | None]: + """增量 chunk 时间戳序列 → (stall_ms, stall_count);chunk <2 → (None, None)。 + + 沉默 = 相邻间隔**严格大于**阈值(恰好 = 阈值不减,pi-tps 口径); + 尾部(末 chunk → 流结束)不计——墙钟基底口径,尾部归活跃侧(设计 §4.2 N1)。 + 无超阈段但 chunk ≥2 时返回 (0.0, 0.0) 而非 None——"观测过且零沉默"与 + "无从观测"是两回事,下游预算公式以 `stall_ms is None` 为分支条件(设计 §4.2 N2)。 + """ + if len(times) < 2: + return None, None + total = 0.0 + count = 0 + # strict=False: 两序列恒差 1(times[1:] 少末尾),B905 要求显式表态——不等长是本设计的本意 + for prev, cur in zip(times, times[1:], strict=False): + gap_ms = (cur - prev) * 1000 + if gap_ms > threshold_ms: + total += gap_ms + count += 1 + return (total, float(count)) if count else (0.0, 0.0) + + async def _iter_sse_deltas( lines: AsyncIterator[str], usage_sink: dict[str, Any] ) -> AsyncIterator[tuple[bool, str]]: @@ -434,6 +457,7 @@ class OpenAICompatTransport: call_id: str, reasoning_effort: Effort | None, first_token_event: asyncio.Event | None, + content_event: asyncio.Event | None, ) -> TransportResult: """一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。 @@ -441,6 +465,8 @@ class OpenAICompatTransport: 在 `_build_payload` 里由 `effective_effort` 裁定,本层只负责把它送到。 `first_token_event` 为对冲处置位: `None` = 不观测首 token;仅流式路径 (`_complete_stream`)在首个增量到达时置位,非流式路径收它但永不置位。 + `content_event` 为中段沉默对冲处置位(1.3.8): 同款约定,流式路径**每个** + 增量 chunk 置位一次,非流式路径收它但永不置位。 """ profile = get_provider(source.provider, registry=self._registry) try: @@ -466,11 +492,11 @@ class OpenAICompatTransport: try: if stream: result = await self._complete_stream( - client, url, payload, source, profile, first_token_event + client, url, payload, source, profile, first_token_event, content_event ) else: result = await self._complete_once( - client, url, payload, source, profile, first_token_event + client, url, payload, source, profile, first_token_event, content_event ) except StreamLivenessTimeout as exc: raise TransientError(f"{source.name} 流活性超时({exc.kind})", **ctx) from exc @@ -558,6 +584,7 @@ class OpenAICompatTransport: source: SourceConfig, profile: ProviderProfile, first_token_event: asyncio.Event | None, + content_event: asyncio.Event | None, ) -> TransportResult: started = time.monotonic() async with client.stream("POST", url, json=payload) as resp: @@ -578,8 +605,17 @@ class OpenAICompatTransport: ttft_ms: float | None = None last = started max_gap = 0.0 + chunk_times: list[float] = [] async for is_content, text in guarded: now = time.monotonic() + # 增量 chunk 时间戳(1.3.8 沉默计量): 循环体只会被真实增量触发—— + # 空 chunk/usage 帧/SSE keepalive 已在 _sse_data_payload/_sse_delta + # 被过滤,这里逐字追加即可;阈值判定不在热路径,派生收敛在 + # _stall_from_chunk_times(设计 §3 派生原则) + chunk_times.append(now) + if content_event is not None: + # 每个增量 chunk 置位一次: 中段沉默对冲的活性复位信号 + content_event.set() if ttft_ms is None: ttft_ms = (now - started) * 1000 # 首个增量即对冲语义上的"首 token"(思考增量同样是存活证据, @@ -595,6 +631,7 @@ class OpenAICompatTransport: self._reject_empty_completion(content, source) prompt, completion, usage_source = _resolve_stream_usage(sink, salvaged) reasoning_tokens = _coerce_reasoning_tokens(sink.get("usage")) + stall_ms, stall_count = _stall_from_chunk_times(chunk_times, source.stall_threshold_ms) return TransportResult( content=content, thinking=thinking, @@ -603,6 +640,8 @@ class OpenAICompatTransport: usage_source=usage_source, ttft_ms=ttft_ms, max_inter_token_ms=(max_gap if ttft_ms is not None else None), + stall_ms=stall_ms, + stall_count=stall_count, raw={"usage": sink.get("usage")}, cached_prompt_tokens=_coerce_cached_tokens(sink.get("usage")), model_reported=_coerce_model_reported(sink.get("model")), @@ -664,11 +703,14 @@ class OpenAICompatTransport: source: SourceConfig, profile: ProviderProfile, first_token_event: asyncio.Event | None, + content_event: asyncio.Event | None, ) -> TransportResult: """非流式快路径(三项目均无,库新增): 单 JSON 响应,仅 total 超时。 接收 `first_token_event` 但**永不置位**: 非流式无中途信号,事件自然退化 - 为纯时间阈值(对冲只能靠 `hedge_after_s` 触发)。 + 为纯时间阈值(对冲只能靠 `hedge_after_s` 触发)。`content_event`(1.3.8) + 同款: 接收但**永不置位**——非流式无"中段"概念,stall 两值走 + TransportResult 默认 None(排除条款)。 """ resp = await client.post(url, json=payload) if resp.status_code != 200: diff --git a/src/polygateway/types.py b/src/polygateway/types.py index a9499aa..536e044 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -585,6 +585,17 @@ class TransportResult: `medium` → 实发 `low`),中间件事后再算一遍必然算成请求档。默认 `None` 是 "未表态/不注入推理参数"(OCR、embedding 等 transport 沉默即此)。""" + stall_ms: float | None = None + """超阈值相邻增量 chunk 沉默总额(1.3.8, issue #27;毫秒,未取整——换算唯一处 + 在 retry 层,设计 §4.2 N8)。 + + 非流式/增量 chunk <2 → `None`("无从观测");观测过但零超阈沉默是 `0.0`, + 两者不得混——下游预算公式以 `stall_ms is None` 为分支条件(设计 §4.2 N2)。 + 尾部(末 chunk → 流结束)不计,归活跃侧(设计 §4.2 N1)。""" + stall_count: float | None = None + """超阈值相邻间隔个数(归因成对报告,Gregg 原则);值域与 `None` 语义同 + `stall_ms`。""" + @dataclass(frozen=True) class SourceConfig: @@ -638,10 +649,19 @@ class SourceConfig: 后者对未知值是 fail-closed(按 `error` 处理),不会替配置兜错,漏判的结果 就是 `EFFORT_FALLBAK` 这种拼写错误静默失效。""" + stall_threshold_ms: int = 1000 + """流式中段沉默的计量阈值(1.3.8, issue #27): 相邻增量 chunk 间隔**严格大于** + 本值才计入 `TransportResult.stall_ms`/`stall_count`。 + + 纯计量口径,不触发任何动作;触发中段对冲的动作阈值是 scope 级 + `HEDGE__STALL_AFTER_S`(设计 §6 两参数分工)。scope 级键 + `STREAM__STALL_THRESHOLD_MS` 经装配层 `dataclasses.replace` 分发进本字段。""" + def __post_init__(self) -> None: self._validate_identity() self._validate_gates() self._validate_watchdog() + self._validate_stall_threshold() self._validate_thinking() self._freeze_extra_body() @@ -680,6 +700,17 @@ class SourceConfig: ): raise ValueError("看门狗不变式要求 0 < inter_token < ttft < timeout_s") + def _validate_stall_threshold(self) -> None: + # bool 是 int 子类: True/False 会被当成 1/0 静默放行,必须显式拒(P5) + if ( + isinstance(self.stall_threshold_ms, bool) + or not isinstance(self.stall_threshold_ms, int) + or self.stall_threshold_ms <= 0 + ): + raise ValueError( + f"SourceConfig.stall_threshold_ms 必须是正整数: {self.stall_threshold_ms!r}" + ) + def _validate_thinking(self) -> None: """推理两键的**归一化**、值域与互不矛盾(issue #20 设计 §4.2)。 diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 5c94c22..bdf1ac8 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -292,6 +292,7 @@ class ObservedTransport: call_id: str, reasoning_effort: Effort | None, first_token_event: asyncio.Event | None, + content_event: asyncio.Event | None, ) -> TransportResult: """与生产端口逐参数同签名。""" with self._capture.attempt_context(call_id): @@ -304,6 +305,7 @@ class ObservedTransport: call_id=call_id, reasoning_effort=reasoning_effort, first_token_event=first_token_event, + content_event=content_event, ) async def embed( diff --git a/tests/integration/test_redis_cross_connection.py b/tests/integration/test_redis_cross_connection.py index ff980ba..fc2f5a2 100644 --- a/tests/integration/test_redis_cross_connection.py +++ b/tests/integration/test_redis_cross_connection.py @@ -76,7 +76,16 @@ class ScriptedTransport: self.entered = asyncio.Event() async def complete( - self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + self, + *, + messages, + source, + stream, + overlay, + call_id, + reasoning_effort, + first_token_event, + content_event, ): self.calls.append(source.name) if self.hang: diff --git a/tests/unit/test_backpressure.py b/tests/unit/test_backpressure.py index 3186e1d..c92581c 100644 --- a/tests/unit/test_backpressure.py +++ b/tests/unit/test_backpressure.py @@ -211,7 +211,16 @@ class ClockAdvancingTransport: self.calls = [] async def complete( - self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + self, + *, + messages, + source, + stream, + overlay, + call_id, + reasoning_effort, + first_token_event, + content_event, ): self.calls.append((source.name, call_id)) advance, action = self.script.pop(0) diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index d2b180f..830548c 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -222,7 +222,16 @@ class _ScriptedGenClockTransport: self.calls = [] async def complete( - self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + self, + *, + messages, + source, + stream, + overlay, + call_id, + reasoning_effort, + first_token_event, + content_event, ): self.calls.append(call_id) result = self._results.pop(0) @@ -1825,7 +1834,16 @@ class _ClockJumpTransport: self.calls = [] async def complete( - self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + self, + *, + messages, + source, + stream, + overlay, + call_id, + reasoning_effort, + first_token_event, + content_event, ): self.calls.append(call_id) self._clock.advance(self._jump) diff --git a/tests/unit/test_hedge.py b/tests/unit/test_hedge.py index d419414..3fafc94 100644 --- a/tests/unit/test_hedge.py +++ b/tests/unit/test_hedge.py @@ -49,14 +49,26 @@ class HedgeTransport: self.calls: list[str] = [] # 逐次记录收到的 first_token_event 身份(H5: 对冲路恒为 None,不再梯次) self.ft_events: list[object] = [] + # content_event 同款逐次记录(1.3.8 N3③: 对冲路恒 None 的断言原料) + self.ce_events: list[object] = [] self.entered = {name: asyncio.Event() for name in scripts} self.release = {name: asyncio.Event() for name in scripts} async def complete( - self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + self, + *, + messages, + source, + stream, + overlay, + call_id, + reasoning_effort, + first_token_event, + content_event, ): self.calls.append(source.name) self.ft_events.append(first_token_event) + self.ce_events.append(content_event) actions = self._scripts[source.name] action = actions.pop(0) if len(actions) > 1 else actions[0] kind = action[0] diff --git a/tests/unit/test_live_evidence.py b/tests/unit/test_live_evidence.py index c58238f..6d6723d 100644 --- a/tests/unit/test_live_evidence.py +++ b/tests/unit/test_live_evidence.py @@ -278,6 +278,7 @@ async def _complete(observed, *, call_id="a", stream=False): call_id=call_id, reasoning_effort=None, first_token_event=None, + content_event=None, ) @@ -1268,6 +1269,7 @@ async def test_structured_first_attempt_requires_exact_initial_messages(): call_id="first", reasoning_effort=None, first_token_event=None, + content_event=None, ) event = capture.attempts(session_id="first", parent_call_id="parent")[0].http[0] assert not request_is_valid(event) diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index 1fcecd8..79772d7 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -25,6 +25,7 @@ from polygateway.transports.openai_compat import ( _iter_sse_deltas, _parse_retry_after, _sse_data_payload, + _stall_from_chunk_times, _translate_429, ) from polygateway.types import ( @@ -81,7 +82,14 @@ def _transport_for(handler, *, registry=None): async def _complete( - transport, source, *, stream=True, overlay=None, reasoning_effort=None, first_token_event=None + transport, + source, + *, + stream=True, + overlay=None, + reasoning_effort=None, + first_token_event=None, + content_event=None, ): return await transport.complete( messages=[{"role": "user", "content": "hi"}], @@ -91,6 +99,7 @@ async def _complete( call_id="cid-1", reasoning_effort=reasoning_effort, first_token_event=first_token_event, + content_event=content_event, ) @@ -277,6 +286,195 @@ class TestFirstTokenEvent: ) +class _CountingEvent(asyncio.Event): + """统计 set() 次数的 Event: content_event "每增量 chunk 置位一次"的观测设施。""" + + def __init__(self) -> None: + super().__init__() + self.set_count = 0 + + def set(self) -> None: + self.set_count += 1 + super().set() + + +def _gated_sse_response(*items): + """事件驱动的 SSE 响应体: str 段直接产出,asyncio.Event 段等待置位。 + + 停顿由测试体用真实 `asyncio.sleep` 制造(计划 §5 真实 loop 钟范式,4× 余量); + 闸门只决定"停顿发生在哪两帧之间",不断定时长,两个方向都不撞窗口。 + """ + + async def gen(): + for item in items: + if isinstance(item, asyncio.Event): + await item.wait() + else: + yield item.encode() + + return httpx.Response(200, content=gen(), headers={"content-type": "text/event-stream"}) + + +class TestContentEvent: + """增量 chunk 处置位(1.3.8, issue #27/#28): 中段沉默计量与对冲的信号源。""" + + async def test_stream_sets_content_event_per_chunk(self): + """流式每个增量 chunk(内容或思考)置位一次——对冲触发窗的沉默复位信号。""" + + def handler(request): + return _sse_stream( + _chunk(reasoning="ponder"), + _chunk(content="hi"), + _chunk(content=" there"), + _chunk(usage=_USAGE), + ) + + event = _CountingEvent() + result = await _complete(_transport_for(handler), _source(), content_event=event) + assert result.content == "hi there" + assert event.set_count == 3 # 3 个增量 chunk 各置位一次 + + async def test_reasoning_chunk_counts_as_liveness(self): + """reasoning_content 增量同样置位并进 chunk 序列(B1: 思考增量是存活证据)。""" + + def handler(request): + return _sse_stream( + _chunk(reasoning="step1"), + _chunk(reasoning="step2"), + _chunk(content="ok"), + _chunk(usage=_USAGE), + ) + + event = _CountingEvent() + result = await _complete(_transport_for(handler), _source(), content_event=event) + assert result.thinking == "step1step2" + assert event.set_count == 3 + # 快速投递下零超阈沉默: "观测过且零沉默"(0.0) ≠ "无从观测"(None); + # 两值为 0.0 而非 None 同时证明 chunk 序列 ≥2(思考增量也入了序列) + assert result.stall_ms == 0.0 and result.stall_count == 0.0 + + async def test_non_stream_never_sets_content_event(self): + """非流式无中途信号: 给了事件也永不置位;stall 两值恒 None(排除条款)。""" + + def handler(request): + return httpx.Response( + 200, json={"choices": [{"message": {"content": "42"}}], "usage": _USAGE} + ) + + event = _CountingEvent() + result = await _complete( + _transport_for(handler), _source(), stream=False, content_event=event + ) + assert result.content == "42" + assert event.set_count == 0 + assert result.stall_ms is None and result.stall_count is None + + async def test_none_content_event_keeps_behavior(self): + """`None` = 调用方不观测增量 chunk(未启用中段对冲): 行为与旧版逐字相同。""" + + def handler(request): + return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE)) + + result = await _complete(_transport_for(handler), _source(), content_event=None) + assert result.content == "ok" + assert result.ttft_ms is not None + + async def test_content_event_is_required_keyword(self): + """端口必填约定同 first_token_event: 漏传必须 TypeError。""" + + def handler(request): + return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE)) + + transport = _transport_for(handler) + with pytest.raises(TypeError): + await transport.complete( + messages=[{"role": "user", "content": "hi"}], + source=_source(), + stream=True, + overlay={}, + call_id="cid-1", + reasoning_effort=None, + first_token_event=None, + ) + + +class TestStallFromChunkTimes: + """沉默计量纯函数(设计 §4.2): 严格大于阈值才减、chunk<2 排除、尾部不计。""" + + def test_gap_above_threshold_accumulates(self): + stall_ms, stall_count = _stall_from_chunk_times([0.0, 0.3, 2.0, 2.1], 1000) + assert stall_ms == pytest.approx(1700.0) # 仅 0.3→2.0 的 1.7s 段超阈 + assert stall_count == 1.0 + + def test_gap_exactly_at_threshold_not_deducted(self): + """间隔恰好 == 阈值不减(严格大于,pi-tps 口径)。""" + assert _stall_from_chunk_times([0.0, 1.0], 1000) == (0.0, 0.0) + + def test_fewer_than_two_chunks_returns_none(self): + """0/1 个 chunk → (None, None): 无从观测,不得伪造 0(vLLM 排除条款)。""" + assert _stall_from_chunk_times([], 1000) == (None, None) + assert _stall_from_chunk_times([0.5], 1000) == (None, None) + + async def test_tail_not_counted(self): + """末 chunk → 流结束(usage/[DONE])的间隔不进 stall(尾部归活跃侧,N1)。""" + gate = asyncio.Event() + + def handler(request): + return _gated_sse_response( + _chunk(content="a"), + _chunk(content="b"), + gate, + _chunk(usage=_USAGE), + "data: [DONE]\n\n", + ) + + async def open_after_pause(): + await asyncio.sleep(0.2) # 真实尾部停顿 200ms(阈值 50ms 的 4×;方向安全: 只会更大) + gate.set() + + task = asyncio.create_task( + _complete(_transport_for(handler), _source(stall_threshold_ms=50), content_event=None) + ) + await asyncio.gather(task, open_after_pause()) + result = task.result() + assert result.content == "ab" + # 唯一的相邻 chunk 间隔是快速段;200ms 的尾部停顿若被计入即超阈暴露 + assert result.stall_ms == 0.0 and result.stall_count == 0.0 + + async def test_empty_or_usage_frames_not_timed(self): + """空 chunk/usage 帧/keepalive 不产生时间戳记录点: 三段停顿只算一次 chunk 间隔。""" + gates = [asyncio.Event() for _ in range(3)] + + def handler(request): + return _gated_sse_response( + _chunk(content="a"), + gates[0], + 'data: {"choices":[{"delta":{}}]}\n\n', # 空 chunk + gates[1], + ": ping\n\n", # SSE keepalive + gates[2], + _chunk(usage=_USAGE), # usage-only 帧 + _chunk(content="b"), + "data: [DONE]\n\n", + ) + + async def open_gates(): + for gate in gates: + await asyncio.sleep(0.2) + gate.set() + + task = asyncio.create_task( + _complete(_transport_for(handler), _source(stall_threshold_ms=50), content_event=None) + ) + await asyncio.gather(task, open_gates()) + result = task.result() + assert result.content == "ab" + # 三个非增量帧若被计时,三段 200ms 间隔各自超阈(count==3); + # 真相: chunk1→chunk2 仅一个间隔(~600ms),count==1 + assert result.stall_count == 1.0 + assert result.stall_ms >= 400.0 # 下界证明停顿真实发生(慢机器只会更大,不设上界) + + class TestMissingDoneSemantics: def _no_done_handler(self, request): return _sse_stream(_chunk(content="partial"), _chunk(usage=_USAGE), done=False) diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index 0eb0efb..819eec0 100644 --- a/tests/unit/test_ports.py +++ b/tests/unit/test_ports.py @@ -73,7 +73,16 @@ class _DummyMw: class _DummyTransport: async def complete( - self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + self, + *, + messages, + source, + stream, + overlay, + call_id, + reasoning_effort, + first_token_event, + content_event, ): raise NotImplementedError diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 3a2f951..62bdc41 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -80,7 +80,16 @@ class FakeTransport: self.entered = asyncio.Event() async def complete( - self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + self, + *, + messages, + source, + stream, + overlay, + call_id, + reasoning_effort, + first_token_event, + content_event, ): self.calls.append((source.name, call_id)) self.efforts.append(reasoning_effort) @@ -110,7 +119,16 @@ class _GenClockTransport: return self._inner.calls async def complete( - self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + self, + *, + messages, + source, + stream, + overlay, + call_id, + reasoning_effort, + first_token_event, + content_event, ): advance = self._advances.pop(0) result = await self._inner.complete( @@ -121,6 +139,7 @@ class _GenClockTransport: call_id=call_id, reasoning_effort=reasoning_effort, first_token_event=first_token_event, + content_event=content_event, ) self._clock.advance(advance) return result diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index d9e1863..55bbc40 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -185,6 +185,19 @@ class TestSourceConfig: _make_source(missing_done="ignore") +def test_source_config_stall_threshold_validation(): + """`stall_threshold_ms`(1.3.8 沉默计量阈值): 默认 1000;0/负/bool → ValueError。""" + assert _make_source().stall_threshold_ms == 1000 + assert _make_source(stall_threshold_ms=50).stall_threshold_ms == 50 + with pytest.raises(ValueError): + _make_source(stall_threshold_ms=0) + with pytest.raises(ValueError): + _make_source(stall_threshold_ms=-5) + # bool 是 int 子类: True 会被当成 1 静默放行,必须显式拒(P5) + with pytest.raises(ValueError): + _make_source(stall_threshold_ms=True) + + class TestEffectiveEstTokens: """TPM 入场预扣量的派生(est_tokens 解耦设计 §2.2)。""" @@ -293,6 +306,12 @@ class TestAuxTypes: assert s.thinking_observation is ThinkingObservation.UNKNOWN +def test_transport_result_stall_fields_default_none(): + """TransportResult 沉默两字段(1.3.8)默认 None;旧位置构造不炸(只增不改)。""" + result = TransportResult("c", "t", 1, 2, "measured", None, None, {}) + assert result.stall_ms is None and result.stall_count is None + + class TestOcrTypes: """M3 OCR 五类型(设计 §3.1): 可构造、frozen、全必填无默认。""" diff --git a/tests/unit/test_usage_source_domain.py b/tests/unit/test_usage_source_domain.py index 61c6b37..152e9fc 100644 --- a/tests/unit/test_usage_source_domain.py +++ b/tests/unit/test_usage_source_domain.py @@ -140,6 +140,7 @@ async def test_salvage_override_stays_in_domain(usage): call_id="cid", reasoning_effort=None, first_token_event=None, + content_event=None, ) assert result.usage_source in USAGE_SOURCES