feat: port three-layer stream liveness watchdog

tests/unit/test_streaming.py

@@ -0,0 +1,108 @@
+"""streaming.py 三层活性看门狗测试(移植自 VT adapters/streaming.py,语义零改动)。"""
+
+import asyncio
+
+import pytest
+
+from polygateway.streaming import StreamLivenessTimeout, stream_with_liveness_timeouts
+
+
+async def _emit(items, *, delay_s=0.0, first_delay_s=None):
+    """可控节奏的假流;first_delay_s 单独控制首项延迟。"""
+    for i, item in enumerate(items):
+        wait = first_delay_s if (i == 0 and first_delay_s is not None) else delay_s
+        if wait:
+            await asyncio.sleep(wait)
+        yield item
+
+
+async def _collect(agen):
+    return [item async for item in agen]
+
+
+class TestPassthrough:
+    async def test_items_pass_through_in_order(self):
+        wrapped = stream_with_liveness_timeouts(
+            _emit(["a", "b", "c"]), ttft_s=1.0, inter_token_s=1.0, total_s=5.0
+        )
+        assert await _collect(wrapped) == ["a", "b", "c"]
+
+    async def test_empty_stream_ends_cleanly(self):
+        wrapped = stream_with_liveness_timeouts(
+            _emit([]), ttft_s=1.0, inter_token_s=1.0, total_s=5.0
+        )
+        assert await _collect(wrapped) == []
+
+
+class TestThreeLayers:
+    async def test_ttft_timeout(self):
+        wrapped = stream_with_liveness_timeouts(
+            _emit(["a"], first_delay_s=0.2), ttft_s=0.05, inter_token_s=1.0, total_s=5.0
+        )
+        with pytest.raises(StreamLivenessTimeout) as ei:
+            await _collect(wrapped)
+        assert ei.value.kind == "ttft"
+        assert ei.value.first_token_seen is False
+
+    async def test_inter_token_timeout(self):
+        wrapped = stream_with_liveness_timeouts(
+            _emit(["a", "b"], delay_s=0.2, first_delay_s=0.0),
+            ttft_s=1.0, inter_token_s=0.05, total_s=5.0,
+        )
+        with pytest.raises(StreamLivenessTimeout) as ei:
+            await _collect(wrapped)
+        assert ei.value.kind == "inter_token"
+        assert ei.value.first_token_seen is True
+
+    async def test_total_timeout(self):
+        # 每项都快,但项数多到总时长超限 → 判 total 而非 inter_token
+        wrapped = stream_with_liveness_timeouts(
+            _emit(list(range(100)), delay_s=0.02), ttft_s=1.0, inter_token_s=1.0, total_s=0.1
+        )
+        with pytest.raises(StreamLivenessTimeout) as ei:
+            await _collect(wrapped)
+        assert ei.value.kind == "total"
+
+
+class TestForeignTimeoutAndCleanup:
+    async def test_upstream_timeout_error_not_reclassified(self):
+        """上游自抛的 TimeoutError 不得被误判为本层活性超时。"""
+
+        async def bad_stream():
+            raise TimeoutError("upstream own timeout")
+            yield  # pragma: no cover
+
+        wrapped = stream_with_liveness_timeouts(
+            bad_stream(), ttft_s=1.0, inter_token_s=1.0, total_s=5.0
+        )
+        with pytest.raises(TimeoutError) as ei:
+            await _collect(wrapped)
+        assert not isinstance(ei.value, StreamLivenessTimeout)
+
+    async def test_underlying_iterator_closed_on_abandon(self):
+        closed = asyncio.Event()
+
+        async def tracked():
+            try:
+                for i in range(100):
+                    await asyncio.sleep(0.01)
+                    yield i
+            finally:
+                closed.set()
+
+        wrapped = stream_with_liveness_timeouts(
+            tracked(), ttft_s=1.0, inter_token_s=1.0, total_s=5.0
+        )
+        assert (await wrapped.__anext__()) == 0
+        await wrapped.aclose()  # 消费方提前放弃 → finally 关闭底层迭代器
+        assert closed.is_set()
+
+    async def test_cancellation_propagates(self):
+        wrapped = stream_with_liveness_timeouts(
+            _emit(["a"], first_delay_s=10.0), ttft_s=30.0, inter_token_s=30.0, total_s=60.0
+        )
+        task = asyncio.ensure_future(_collect(wrapped))
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task