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/

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