feat: add a required first-token event to the transport port

Transport.complete gains the keyword-only first_token_event (no default,
per the port convention): streaming sets it on the first delta, the
non-streaming path accepts it but never sets it, None means the caller
does not observe the first token. All fake/wrapping transports and the
three direct call sites follow the signature; the e2e wrapper forwards.

Red-green evidence: tests/outputs/137/t1/ (batch A TypeError red, then
147 file tests + 1550 unit tests green).

src/polygateway/middleware/retry.py

@@ -297,6 +297,8 @@ class RetryMW:
                 # 逐次尝试原样重传: 换源不改变调用方要的档位(源级默认由 transport
                 # 自己按选中的源解析,两者在 effective_effort 里汇合)
                 reasoning_effort=request.reasoning_effort,
+                # T3 对冲编排接线前恒为 None: 调用方不观测首 token(计划 §3.1)
+                first_token_event=None,
             )
             if result.usage_source == "unavailable":
                 # 用量不可得时按入场预扣量结算(delta==0),否则押金会被整笔退回,

src/polygateway/ports.py

@@ -5,6 +5,7 @@
 时间量纲一律**秒**(CHS Redis 实现内部的毫秒换算是后端私事,不进契约)。
 """
 
+import asyncio
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from enum import StrEnum
@@ -46,6 +47,10 @@ class Transport(Protocol):
     该参数**不设默认值**,与 `TelemetryRecorder.record_llm_call` 同一既有约定:
     库外无第三方实现者,写全签名的成本为零,而默认值会把"某一层漏传"变成静默的
     "调用方没表态"——一次本该报错的漏配就此变成一次悄悄涨价的调用。
+
+    `first_token_event` 同一约定(1.3.7 对冲 H2): `None` = 调用方不观测首 token
+    (未启用对冲);非流式实现**永不置位**(物理上无中途信号,事件自然退化为纯时间
+    阈值),流式实现在首个增量(内容或思考)到达时置位。
     """
 
     async def complete(
@@ -57,6 +62,7 @@ class Transport(Protocol):
         overlay: dict[str, Any],
         call_id: str,
         reasoning_effort: Effort | None,
+        first_token_event: asyncio.Event | None,
     ) -> TransportResult: ...
 
 

src/polygateway/transports/openai_compat.py

@@ -46,6 +46,7 @@ from polygateway.types import (
 )
 
 if TYPE_CHECKING:
+    import asyncio
     from collections.abc import AsyncIterator, Callable, Mapping
 
 _THINK_PATTERN = re.compile(r"<think>(.*?)</think>", re.DOTALL)
@@ -432,11 +433,14 @@ class OpenAICompatTransport:
         overlay: dict[str, Any],
         call_id: str,
         reasoning_effort: Effort | None,
+        first_token_event: asyncio.Event | None,
     ) -> TransportResult:
         """一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。
 
         `reasoning_effort` 是**请求级**档位(`None` = 不表态);它与源级配置的优先级
         在 `_build_payload` 里由 `effective_effort` 裁定,本层只负责把它送到。
+        `first_token_event` 为对冲处置位: `None` = 不观测首 token;仅流式路径
+        (`_complete_stream`)在首个增量到达时置位,非流式路径收它但永不置位。
         """
         profile = get_provider(source.provider, registry=self._registry)
         try:
@@ -461,9 +465,13 @@ class OpenAICompatTransport:
         ctx: dict[str, Any] = {"source_name": source.name, "operation": "chat"}
         try:
             if stream:
-                result = await self._complete_stream(client, url, payload, source, profile)
+                result = await self._complete_stream(
+                    client, url, payload, source, profile, first_token_event
+                )
             else:
-                result = await self._complete_once(client, url, payload, source, profile)
+                result = await self._complete_once(
+                    client, url, payload, source, profile, first_token_event
+                )
         except StreamLivenessTimeout as exc:
             raise TransientError(f"{source.name} 流活性超时({exc.kind})", **ctx) from exc
         except httpx.TimeoutException as exc:
@@ -549,6 +557,7 @@ class OpenAICompatTransport:
         payload: dict[str, Any],
         source: SourceConfig,
         profile: ProviderProfile,
+        first_token_event: asyncio.Event | None,
     ) -> TransportResult:
         started = time.monotonic()
         async with client.stream("POST", url, json=payload) as resp:
@@ -573,6 +582,10 @@ class OpenAICompatTransport:
                 now = time.monotonic()
                 if ttft_ms is None:
                     ttft_ms = (now - started) * 1000
+                    # 首个增量即对冲语义上的"首 token"(思考增量同样是存活证据,
+                    # 与看门狗活性口径一致);None = 调用方未启用对冲,零分支成本
+                    if first_token_event is not None:
+                        first_token_event.set()
                 else:
                     max_gap = max(max_gap, (now - last) * 1000)
                 last = now
@@ -650,8 +663,13 @@ class OpenAICompatTransport:
         payload: dict[str, Any],
         source: SourceConfig,
         profile: ProviderProfile,
+        first_token_event: asyncio.Event | None,
     ) -> TransportResult:
-        """非流式快路径(三项目均无,库新增): 单 JSON 响应,仅 total 超时。"""
+        """非流式快路径(三项目均无,库新增): 单 JSON 响应,仅 total 超时。
+
+        接收 `first_token_event` 但**永不置位**: 非流式无中途信号,事件自然退化
+        为纯时间阈值(对冲只能靠 `hedge_after_s` 触发)。
+        """
         resp = await client.post(url, json=payload)
         if resp.status_code != 200:
             raise _status_to_error(

tests/e2e/conftest.py

@@ -1,5 +1,6 @@
 """测试侧独立 HTTP 取证装配；无环境自读取或成功 SSE 预读。"""
 
+import asyncio
 from collections.abc import AsyncIterator, Iterator, Mapping
 from contextlib import AsyncExitStack, asynccontextmanager, contextmanager
 from contextvars import ContextVar
@@ -290,6 +291,7 @@ class ObservedTransport:
         overlay: dict[str, Any],
         call_id: str,
         reasoning_effort: Effort | None,
+        first_token_event: asyncio.Event | None,
     ) -> TransportResult:
         """与生产端口逐参数同签名。"""
         with self._capture.attempt_context(call_id):
@@ -301,6 +303,7 @@ class ObservedTransport:
                 overlay=overlay,
                 call_id=call_id,
                 reasoning_effort=reasoning_effort,
+                first_token_event=first_token_event,
             )
 
     async def embed(

tests/integration/test_redis_cross_connection.py

@@ -75,7 +75,9 @@ class ScriptedTransport:
         # 取消用例的确定性窗口(同 test_retry FakeTransport): 进入挂起即置位
         self.entered = asyncio.Event()
 
-    async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort):
+    async def complete(
+        self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
+    ):
         self.calls.append(source.name)
         if self.hang:
             self.entered.set()

tests/unit/test_backpressure.py

@@ -210,7 +210,9 @@ class ClockAdvancingTransport:
         self.clock = clock
         self.calls = []
 
-    async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort):
+    async def complete(
+        self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
+    ):
         self.calls.append((source.name, call_id))
         advance, action = self.script.pop(0)
         self.clock.advance(advance)

tests/unit/test_client.py

@@ -1738,7 +1738,9 @@ class _ClockJumpTransport:
         self._jump = jump
         self.calls = []
 
-    async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort):
+    async def complete(
+        self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
+    ):
         self.calls.append(call_id)
         self._clock.advance(self._jump)
         return _ok()

tests/unit/test_live_evidence.py

@@ -277,6 +277,7 @@ async def _complete(observed, *, call_id="a", stream=False):
         overlay={},
         call_id=call_id,
         reasoning_effort=None,
+        first_token_event=None,
     )
 
 
@@ -1266,6 +1267,7 @@ async def test_structured_first_attempt_requires_exact_initial_messages():
                 overlay={},
                 call_id="first",
                 reasoning_effort=None,
+                first_token_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

@@ -3,6 +3,7 @@
 SSE 帧样本按三项目真实网关响应形态二次构造(OpenAI 兼容 chunk 结构)。
 """
 
+import asyncio
 import json
 
 import httpx
@@ -79,7 +80,9 @@ def _transport_for(handler, *, registry=None):
     )
 
 
-async def _complete(transport, source, *, stream=True, overlay=None, reasoning_effort=None):
+async def _complete(
+    transport, source, *, stream=True, overlay=None, reasoning_effort=None, first_token_event=None
+):
     return await transport.complete(
         messages=[{"role": "user", "content": "hi"}],
         source=source,
@@ -87,6 +90,7 @@ async def _complete(transport, source, *, stream=True, overlay=None, reasoning_e
         overlay=overlay or {},
         call_id="cid-1",
         reasoning_effort=reasoning_effort,
+        first_token_event=first_token_event,
     )
 
 
@@ -214,6 +218,65 @@ class TestStreamHappyPath:
         assert await _recorded_cost(result, source) is None
 
 
+class TestFirstTokenEvent:
+    """首 token 处置位(1.3.7 对冲 H2): 流式置位、非流式永不置位、None 不观测。"""
+
+    async def test_stream_sets_first_token_event(self):
+        """流式首 token(内容或思考增量)到达即置位——对冲触发窗的取消信号。"""
+
+        def handler(request):
+            return _sse_stream(
+                _chunk(reasoning="ponder"), _chunk(content="hi"), _chunk(usage=_USAGE)
+            )
+
+        event = asyncio.Event()
+        result = await _complete(_transport_for(handler), _source(), first_token_event=event)
+        assert result.content == "hi"
+        assert event.is_set()
+
+    async def test_non_stream_never_sets_first_token_event(self):
+        """非流式物理上无中途信号: 即使调用方给了事件,本路径也永不置位。"""
+
+        def handler(request):
+            return httpx.Response(
+                200, json={"choices": [{"message": {"content": "42"}}], "usage": _USAGE}
+            )
+
+        event = asyncio.Event()
+        result = await _complete(
+            _transport_for(handler), _source(), stream=False, first_token_event=event
+        )
+        assert result.content == "42"
+        assert not event.is_set()
+
+    async def test_none_first_token_event_keeps_behavior(self):
+        """`None` = 调用方不观测首 token(未启用对冲): 行为与旧版逐字相同。"""
+
+        def handler(request):
+            return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))
+
+        result = await _complete(_transport_for(handler), _source(), first_token_event=None)
+        assert result.content == "ok"
+        assert result.ttft_ms is not None
+
+    async def test_first_token_event_is_required_keyword(self):
+        """端口必填约定: 漏传必须 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,
+            )
+
+
 class TestMissingDoneSemantics:
     def _no_done_handler(self, request):
         return _sse_stream(_chunk(content="partial"), _chunk(usage=_USAGE), done=False)

tests/unit/test_ports.py

@@ -72,7 +72,9 @@ class _DummyMw:
 
 
 class _DummyTransport:
-    async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort):
+    async def complete(
+        self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
+    ):
         raise NotImplementedError
 
 

tests/unit/test_retry.py

@@ -79,7 +79,9 @@ class FakeTransport:
         # 取消用例的确定性窗口: 进入 hang 分支即置位, 用例据此取消而非 sleep 猜时长
         self.entered = asyncio.Event()
 
-    async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort):
+    async def complete(
+        self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
+    ):
         self.calls.append((source.name, call_id))
         self.efforts.append(reasoning_effort)
         action = self.script.pop(0)

tests/unit/test_usage_source_domain.py

@@ -139,6 +139,7 @@ async def test_salvage_override_stays_in_domain(usage):
         overlay={},
         call_id="cid",
         reasoning_effort=None,
+        first_token_event=None,
     )
     assert result.usage_source in USAGE_SOURCES