From 0a6d6225db835c64266a6f4c6bdcd314e0914fba Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 10 Sep 2026 13:16:38 -0400 Subject: [PATCH] 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 | 2 + src/polygateway/ports.py | 6 ++ src/polygateway/transports/openai_compat.py | 24 ++++++- tests/e2e/conftest.py | 3 + .../test_redis_cross_connection.py | 4 +- tests/unit/test_backpressure.py | 4 +- tests/unit/test_client.py | 4 +- tests/unit/test_live_evidence.py | 2 + tests/unit/test_openai_compat.py | 65 ++++++++++++++++++- tests/unit/test_ports.py | 4 +- tests/unit/test_retry.py | 4 +- tests/unit/test_usage_source_domain.py | 1 + 12 files changed, 114 insertions(+), 9 deletions(-) diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 9e24d96..9a8a782 100644 --- a/src/polygateway/middleware/retry.py +++ b/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),否则押金会被整笔退回, diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index a8397d8..7b595a1 100644 --- a/src/polygateway/ports.py +++ b/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: ... diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index e09572c..554141b 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/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"(.*?)", 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( diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index c5a65cd..5c94c22 100644 --- a/tests/e2e/conftest.py +++ b/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( diff --git a/tests/integration/test_redis_cross_connection.py b/tests/integration/test_redis_cross_connection.py index 5405154..ff980ba 100644 --- a/tests/integration/test_redis_cross_connection.py +++ b/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() diff --git a/tests/unit/test_backpressure.py b/tests/unit/test_backpressure.py index 479ea05..3186e1d 100644 --- a/tests/unit/test_backpressure.py +++ b/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) diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 5d8cc80..47f5b0e 100644 --- a/tests/unit/test_client.py +++ b/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() diff --git a/tests/unit/test_live_evidence.py b/tests/unit/test_live_evidence.py index 1c75e2d..c58238f 100644 --- a/tests/unit/test_live_evidence.py +++ b/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) diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index 47102aa..1fcecd8 100644 --- a/tests/unit/test_openai_compat.py +++ b/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) diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index 57beecc..0eb0efb 100644 --- a/tests/unit/test_ports.py +++ b/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 diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 214b487..309502f 100644 --- a/tests/unit/test_retry.py +++ b/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) diff --git a/tests/unit/test_usage_source_domain.py b/tests/unit/test_usage_source_domain.py index 36fff7a..61c6b37 100644 --- a/tests/unit/test_usage_source_domain.py +++ b/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