feat: expose bare generation time and hedge flags in CallStats

CallStats gains hedges/generation_ms/hedge_won (all defaulted, appended
after total_latency_ms); _CallContext counts them via record_generation
(overwrite for chat/OCR, accumulate for embedding batches) and
register_hedge, and snapshot carries them out. All three _attempt
implementations time only the transport call itself on the same injected
clock as total_latency_ms; the chat sink is recorded by the orchestrator
so hedge winner attribution stays with T3. Hedge counters stay 0/False
until the T3 orchestration lands.

Red-green evidence: tests/outputs/137/t2/ (10 new tests AttributeError
red, then green; full unit+contracts 1621 passed).

tests/unit/test_client.py

@@ -144,6 +144,66 @@ class TestChatEndToEnd:
                 await client.chat([{"role": "user", "content": "hi"}], structured="json")
 
 
+class TestGenerationMsClient:
+    """裸生成时间的 client 级口径(1.3.7 批次 C2/F)。
+
+    `_ScriptedGenClockTransport` 在每次 transport 调用内推进注入钟,
+    使"时间花在哪"可断言(同 test_backpressure.ClockAdvancingTransport 范式)。
+    """
+
+    _MSG = [{"role": "user", "content": "hi"}]
+
+    async def test_generation_ms_structured_last_round_wins(self):
+        """结构化重问覆盖而非累加: 首轮坏 JSON 推进 1s,重问轮推进 0.25s → 250。
+
+        推进量取二进制可精确表示值: int 截断下非精确值会因浮点误差少 1ms。
+        """
+        from pydantic import BaseModel
+
+        class Answer(BaseModel):
+            answer: int
+
+        clock = _StatsClock()
+        transport = _ScriptedGenClockTransport(
+            [_ok("not json at all"), _ok('{"answer": 1}')], [1.0, 0.25], clock
+        )
+        async with _client(transport=transport, structured_max_retries=1, now=clock) as client:
+            resp = await client.chat(self._MSG, structured=Answer)
+        assert resp.content == '{"answer": 1}' and len(transport.calls) == 2
+        assert resp.call_stats is not None
+        assert resp.call_stats.attempts == 2
+        assert resp.call_stats.generation_ms == 250
+
+    async def test_cache_hit_generation_ms_zero(self):
+        """缓存命中不产生 transport 调用: generation_ms 恒 0(0 是实测,非"未知")。"""
+        client = _client(cache=InMemoryCache(), cache_namespace="proj", cache_ttl_s=3600)
+        async with client:
+            first = await client.chat(self._MSG)
+            second = await client.chat(self._MSG)
+        assert first.cache_hit is False and second.cache_hit is True
+        assert second.call_stats is not None
+        assert second.call_stats.generation_ms == 0
+        assert second.call_stats.attempts == 0
+
+
+class _ScriptedGenClockTransport:
+    """脚本化假 transport: 每次成功调用在返回前按脚本推进注入钟(批次 C2/F)。"""
+
+    def __init__(self, results, advances, clock):
+        self._results = list(results)
+        self._advances = list(advances)
+        self._clock = clock
+        self.calls = []
+
+    async def complete(
+        self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
+    ):
+        self.calls.append(call_id)
+        result = self._results.pop(0)
+        self._clock.advance(self._advances.pop(0))
+        return result
+
+
 class TestSamplingOverlay:
     """调用级采样参数入口(issue #4 Task 3)。"""
 

tests/unit/test_embedding.py

@@ -319,6 +319,20 @@ class TestEmbedBatching:
         assert resp.cost is None
 
 
+class TestEmbedGenerationMs:
+    """裸生成时间按批累加(1.3.7 H8): generation_ms = 各批 transport 耗时之和。"""
+
+    async def test_generation_ms_sums_batch_transports(self):
+        # 0.25s 为二进制可精确表示值: int 截断下非精确值会因浮点误差少 1ms
+        clock = FakeClock()
+        transport = _ClockAdvancingEmbedTransport([(0.25, "ok"), (0.25, "ok")], clock)
+        client, _ = _embed_client([_src()], [], transport=transport, now=clock)
+        resp = await client.embed(["a", "bb", "ccc", "dddd"])
+        assert resp.call_stats is not None
+        assert resp.call_stats.attempts == 2
+        assert resp.call_stats.generation_ms == 500
+
+
 class TestEmbedPostProcess:
     async def test_normalize_l2(self):
         raw = EmbeddingTransportResult(

tests/unit/test_ocr_client.py

@@ -196,6 +196,21 @@ class TestSuccessPaths:
             await client.parse_layout(b"")
 
 
+class TestOcrGenerationMs:
+    """OCR 裸生成时间单次覆盖(1.3.7 H8): 计时只包 transport 调用本身。"""
+
+    async def test_generation_ms_single_transport_call(self):
+        # 0.5s 为二进制可精确表示值: int 截断下非精确值会因浮点误差少 1ms
+        clock = FakeClock()
+        transport = ClockAdvancingOcrTransport([(0.5, "text")], clock)
+        client, _, _ = _client([_src()], [], now=clock, transport=transport)
+        r = await client.recognize_text(b"jpg")
+        assert r.text == "LINE-1"
+        assert r.call_stats is not None
+        assert r.call_stats.attempts == 1
+        assert r.call_stats.generation_ms == 500
+
+
 class TestFailover:
     async def test_transient_retries_with_backoff(self):
         sleeps = []

tests/unit/test_retry.py

@@ -93,6 +93,39 @@ class FakeTransport:
         return action
 
 
+class _GenClockTransport:
+    """委托 FakeTransport 的薄包装: 每次调用返回前按脚本推进注入钟(1.3.7 批次 C)。
+
+    generation_ms 的口径是"只计 transport 调用本身",故推进必须发生在被包
+    transport 内部;退避耗时由用例自带的 sleep 闭包推进,与本包装无关。
+    """
+
+    def __init__(self, script, advances, clock):
+        self._inner = FakeTransport(script)
+        self._advances = list(advances)
+        self._clock = clock
+
+    @property
+    def calls(self):
+        return self._inner.calls
+
+    async def complete(
+        self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
+    ):
+        advance = self._advances.pop(0)
+        result = await self._inner.complete(
+            messages=messages,
+            source=source,
+            stream=stream,
+            overlay=overlay,
+            call_id=call_id,
+            reasoning_effort=reasoning_effort,
+            first_token_event=first_token_event,
+        )
+        self._clock.advance(advance)
+        return result
+
+
 class HangingGate(InMemoryGate):
     """在指定记账写回处永久挂起的门控: 把"取消落在某个 await 上"变成确定性事件。
 
@@ -141,6 +174,8 @@ def _harness(
     selector=None,
     pacer=None,
     gate=None,
+    transport=None,
+    sleep=None,
 ):
     clock = clock or FakeClock()
     limiter = InMemoryLimiter(
@@ -151,8 +186,8 @@ def _harness(
         now=clock,
     )
     gate = gate if gate is not None else InMemoryGate(config=_BREAKER, now=clock)
-    transport = FakeTransport(script)
-    sleep = FakeSleep()
+    transport = transport if transport is not None else FakeTransport(script)
+    sleep = sleep if sleep is not None else FakeSleep()
     mw = RetryMW(
         scope="llm",
         sources=sources,
@@ -916,6 +951,54 @@ class TestRateLimitPushback:
         assert len(transport.calls) == 3
 
 
+class TestGenerationMs:
+    """裸生成时间(1.3.7 H8): 只计 transport 调用本身,不含退避/准入/遥测收尾。"""
+
+    def _ctx(self, clock):
+        from polygateway.types import _CallContext
+
+        return _CallContext(now=clock)
+
+    async def test_generation_ms_excludes_backoff_and_admission(self):
+        """[Transient, ok] 脚本: 退避推进 5s、成功次 transport 推进 0.25s。
+
+        generation_ms 恒等于成功次 transport 的 250ms;若口径混入了退避,
+        它会涨到 5250ms 量级——与 total_latency_ms 的下界断言互为对偶。
+        推进量取二进制可精确表示值(0.25/5.0): int 截断下 0.2 之类会因浮点
+        误差落到 199,断言随之抖动(同 test_types 既有用例只用 1.5/2.0 的惯例)。
+        """
+        clock = FakeClock()
+
+        async def advancing_sleep(seconds):
+            clock.advance(seconds)
+
+        transport = _GenClockTransport(
+            [TransientError("boom", source_name="a"), _ok()], [0.0, 0.25], clock
+        )
+        mw, *_ = _harness(
+            [_src("a")],
+            [],
+            clock=clock,
+            transport=transport,
+            sleep=advancing_sleep,
+            rng=lambda: 2.0,  # backoff = 2.0 * (0.5 + 2.0) = 5.0s
+        )
+        ctx = self._ctx(clock)
+        resp = await mw(dataclasses.replace(_REQ, call_context=ctx))
+        assert resp.content == "ok" and len(transport.calls) == 2
+        stats = ctx.snapshot()
+        assert stats.generation_ms == 250
+        assert stats.total_latency_ms >= 5250
+
+    async def test_generation_ms_zero_hedge_flags_without_hedging(self):
+        """无对冲时 hedges/hedge_won 恒 0/False(对冲登记是 T3 的事)。"""
+        mw, *_ = _harness([_src("a")], [_ok()])
+        ctx = self._ctx(FakeClock())
+        await mw(dataclasses.replace(_REQ, call_context=ctx))
+        stats = ctx.snapshot()
+        assert stats.hedges == 0 and stats.hedge_won is False
+
+
 class TestLogicalAttemptCounting:
     """尝试登记在 transport 调用**之前**(1.3.5 设计 §4)。
 

tests/unit/test_types.py

@@ -622,6 +622,55 @@ class TestCallStatsAndContext:
         with pytest.raises(dataclasses.FrozenInstanceError):
             stats.attempts = 3
 
+    def test_callstats_hedge_fields_default(self):
+        """1.3.7 三字段全带默认值: 仅旧三参数构造不炸,无对冲恒 0/0/False。"""
+        from polygateway.types import CallStats
+
+        stats = CallStats(logical_call_id="lc-1", attempts=2, total_latency_ms=15)
+        assert stats.hedges == 0
+        assert stats.generation_ms == 0
+        assert stats.hedge_won is False
+        explicit = CallStats(
+            logical_call_id="lc-2",
+            attempts=2,
+            total_latency_ms=15,
+            hedges=1,
+            generation_ms=42,
+            hedge_won=True,
+        )
+        assert (explicit.hedges, explicit.generation_ms, explicit.hedge_won) == (1, 42, True)
+
+    def test_callcontext_record_generation_overwrite_and_accumulate(self):
+        """chat/OCR 覆盖(结构化重问最后一轮为准);embedding 分批累加。"""
+        from polygateway.types import _CallContext
+
+        ctx = _CallContext(now=_FakeMonotonic())
+        ctx.record_generation(100, accumulate=False)
+        ctx.record_generation(30, accumulate=False)
+        assert ctx.snapshot().generation_ms == 30
+        ctx.record_generation(50, accumulate=True)
+        assert ctx.snapshot().generation_ms == 80
+
+    def test_callcontext_register_hedge_counts(self):
+        """对冲路实际发出即计数;赢家裁定后一次性登记赢家身份。"""
+        from polygateway.types import _CallContext
+
+        ctx = _CallContext(now=_FakeMonotonic())
+        ctx.register_hedge(hedge_won=False)
+        ctx.register_hedge(hedge_won=True)
+        stats = ctx.snapshot()
+        assert stats.hedges == 2 and stats.hedge_won is True
+
+    def test_snapshot_includes_hedge_fields(self):
+        """快照把三字段带出: 裸生成时间与对冲计数不停留在内部状态里。"""
+        from polygateway.types import _CallContext
+
+        ctx = _CallContext(now=_FakeMonotonic())
+        ctx.record_generation(200, accumulate=False)
+        ctx.register_hedge(hedge_won=True)
+        stats = ctx.snapshot()
+        assert (stats.hedges, stats.generation_ms, stats.hedge_won) == (1, 200, True)
+
     def test_context_counts_attempts_and_freezes_elapsed(self):
         """快照是同步冻结的时间切片: 登记两次尝试后耗时按注入钟折算成毫秒。"""
         from polygateway.types import _CallContext