feat: add a content event and stall accounting to the transport port

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

@@ -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,
                 )
             )

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: ...
 
 

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:

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)。
 

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(

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:

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)

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)

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]

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)

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)

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
 

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

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、全必填无默认。"""
 

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