feat: redefine generation_ms as net generation time with stall attribution

CallStats gains stall_ms/stall_count (None = no stall observation);
winner sample triple (wall, stall_ms, stall_count) converted once in
_attempt success branch; net = max(0, wall - stall) converged into the
shared _record_winner_sample used by both the single-path guard and
hedge winner registration. Two-fail hedge rounds record nothing
(no winner to attribute). Design section 4.2; issues #27.

src/polygateway/middleware/retry.py

@@ -58,6 +58,7 @@ if TYPE_CHECKING:
         RetryPolicy,
         SourceConfig,
         TransportResult,
+        _CallContext,
     )
 
 
@@ -165,6 +166,11 @@ def _is_rate_limited(outcome: LLMResponse | _Failed) -> bool:
 
 
 _HEDGE_LOSER_ATTR = "_polygateway_hedge_loser"
+
+# 生成样本三元组(1.3.8 设计 §4.2 N8): (墙钟 ms, 沉默总额 ms, 超阈值沉默段数)。
+# 取整换算全库只在 _attempt 成功分支发生一次,消费侧(_record_generation/
+# _register_hedge)只做净口径换算与透传,不再取整。
+_GenerationSample = tuple[int, int | None, int | None]
 """编排在 cancel() 之前给输家任务置位的标记;_attempt 读它选遥测标签。"""
 
 
@@ -262,9 +268,10 @@ class RetryMW:
                 await self._admission.on_no_runnable(gate_rejections, reasons, clock)
                 continue
             async with clock.attempting() as attempt:
-                # 每轮一个 sink: 裸生成时间由编排裁定归属(T2 单路 = 成功那轮;
-                # T3 对冲 = 赢家那一路),`_attempt` 只负责把本次 transport 耗时投进来
-                generation_sink: list[int] = []
+                # 每轮一个 sink: 生成样本(墙钟+沉默归因)由编排裁定归属
+                # (单路 = 成功那轮;对冲 = 赢家那一路),`_attempt` 只负责把本次
+                # transport 耗时与沉默观测投进来
+                generation_sink: list[_GenerationSample] = []
                 if self._hedge_after_s is None:
                     # 默认关闭: 逐字 1.3.6 单路路径(默认关闭回归门据此成立)
                     outcome = await self._attempt(
@@ -284,10 +291,9 @@ class RetryMW:
                     # 免了重试预算就得进 stall 账,否则这段耗时无人治理(见 StallClock)
                     attempt.refund()
             if isinstance(outcome, LLMResponse):
-                # 与 register_attempt 同款 None 守卫: 库内现场构造的请求跳过登记;
                 # 对冲轮由 _attempt_hedged 按赢家归属登记,此处外层 sink 恒为空
-                if request.call_context is not None and generation_sink:
-                    request.call_context.record_generation(generation_sink[0], accumulate=False)
+                # (守卫在 _record_generation 内: 上下文缺失/空 sink 均跳过)
+                self._record_generation(request, generation_sink)
                 return outcome
             if not rate_limited:
                 fails += 1
@@ -314,7 +320,7 @@ class RetryMW:
         *,
         first_token_event: asyncio.Event | None,
         content_event: asyncio.Event | None,
-        generation_sink: list[int],
+        generation_sink: list[_GenerationSample],
     ) -> LLMResponse | _Failed:
         call_id = str(uuid.uuid4())
         started = self._now()
@@ -328,8 +334,9 @@ class RetryMW:
         if request.call_context is not None:
             request.call_context.register_attempt()
         try:
-            # 裸生成时间(1.3.7 H8): 计时只包 transport 调用本身,起点紧贴调用前、
-            # 终点为返回后首句(中间无 await);取消落进来时 transport 未返回,不计。
+            # 净生成时间(1.3.8): 此处只量 transport 墙钟(起点紧贴调用前、
+            # 终点为返回后首句,中间无 await),沉默扣减在消费侧统一换算;
+            # 取消落进来时 transport 未返回,不计。
             # 与该链路 total_latency_ms 同一只注入钟,差值(波动开销)才有意义
             gen_started = self._now()
             result = await self._transport.complete(
@@ -348,7 +355,12 @@ class RetryMW:
                 # T4 才在启用 stall 对冲的流式原路上接真事件(设计 §5.2 N3③)
                 content_event=content_event,
             )
-            generation_sink.append(int((self._now() - gen_started) * 1000))
+            # 取整全库唯一处(设计 §4.2 N8): transport 的 float 沉默值在此
+            # 一次性转 int;None(非流式/增量 chunk<2)透传,消费侧不退减
+            wall_ms = int((self._now() - gen_started) * 1000)
+            stall_ms_i = None if result.stall_ms is None else int(result.stall_ms)
+            stall_count_i = None if result.stall_count is None else int(result.stall_count)
+            generation_sink.append((wall_ms, stall_ms_i, stall_count_i))
             if result.usage_source == "unavailable":
                 # 用量不可得时按入场预扣量结算(delta==0),否则押金会被整笔退回,
                 # 对"从不返回 usage 帧"的源等于 TPM 闸失效(设计 §3.2 #9)
@@ -429,7 +441,7 @@ class RetryMW:
         # Phase 1 启动原路: 事件与 sink 每轮新建(局部状态,严禁实例属性)
         source, permit, entry = picked
         first_token: asyncio.Event = asyncio.Event()
-        sink_p: list[int] = []
+        sink_p: list[_GenerationSample] = []
         primary = asyncio.create_task(
             self._attempt(
                 request,
@@ -480,7 +492,7 @@ class RetryMW:
                 return outcome
             # Phase 4 启动对冲路(v1 单路,H5): 对冲路恒传 first_token_event=None,
             # 不再触发梯次对冲;content_event 同恒 None(N3③,v1 不梯次)
-            sink_h: list[int] = []
+            sink_h: list[_GenerationSample] = []
             hedge = asyncio.create_task(
                 self._attempt(
                     request,
@@ -579,14 +591,27 @@ class RetryMW:
         )
 
     @staticmethod
-    def _record_generation(request: ChatRequest, sink: list[int]) -> None:
+    def _record_generation(request: ChatRequest, sink: list[_GenerationSample]) -> None:
         """record_generation 的共享守卫: 上下文缺失(库内现场构造)或空 sink 均跳过。"""
         if request.call_context is not None and sink:
-            request.call_context.record_generation(sink[0], accumulate=False)
+            RetryMW._record_winner_sample(request.call_context, sink[0])
+
+    @staticmethod
+    def _record_winner_sample(context: _CallContext, sample: _GenerationSample) -> None:
+        """净口径唯一换算(设计 §4.2): net = max(0, 墙钟 − 沉默),clamp 防御(N4)。
+
+        stall 两值透传: None(非流式/增量 chunk<2)时 net=墙钟不退减、快照归因
+        保持 None(排除条款——None 透传天然达成,无需分支)。
+        """
+        wall_ms, stall_ms, stall_count = sample
+        net_ms = max(0, wall_ms - (stall_ms or 0))
+        context.record_generation(
+            net_ms, accumulate=False, stall_ms=stall_ms, stall_count=stall_count
+        )
 
     @staticmethod
     def _register_hedge(
-        request: ChatRequest, *, hedge_won: bool, winner_sink: list[int] | None
+        request: ChatRequest, *, hedge_won: bool, winner_sink: list[_GenerationSample] | None
     ) -> None:
         """对冲登记(赢家裁定后一次性;同 register_attempt 的 None 守卫)。
 
@@ -597,7 +622,7 @@ class RetryMW:
         if context is None:
             return
         if winner_sink:
-            context.record_generation(winner_sink[0], accumulate=False)
+            RetryMW._record_winner_sample(context, winner_sink[0])
         context.register_hedge(hedge_won=hedge_won)
 
     async def _on_rejected(

src/polygateway/types.py

@@ -320,9 +320,20 @@ class CallStats:
     hedges: int = 0
     """本次逻辑调用实际并发发出的对冲路数(触发但准入失败静默不计);1.3.6 及以前恒 0。"""
     generation_ms: int = 0
-    """裸生成时间: 赢家/成功那次 transport 调用的墙钟时长(口径见设计 §4.5 H8)。"""
+    """净生成时间(1.3.8 起重定义,设计 §4.2): 赢家/成功那次 transport 墙钟
+    − Σ(超阈值中段沉默);仍含 TTFT 与尾部(末 chunk→流结束)。
+
+    非流式/增量 chunk<2 无沉默观测、不退减(= 原墙钟,此时 stall 两值 None)。
+    墙钟还原 = generation_ms + stall_ms。1.3.7 及以前本字段是完整墙钟,
+    口径变化见 CHANGELOG 行为变更段。"""
     hedge_won: bool = False
     """赢家是否为对冲路;无对冲恒 False。"""
+    stall_ms: int | None = None
+    """赢家路超阈值中段沉默总额(净口径归因;None = 非流式/chunk<2/无赢家)。
+
+    墙钟 = generation_ms + stall_ms;None 是"无沉默观测",不是 0。"""
+    stall_count: int | None = None
+    """超阈值相邻间隔个数(归因成对报告,Gregg 原则,调研 §B2)。"""
 
 
 class _CallContext:
@@ -340,6 +351,8 @@ class _CallContext:
         "_hedge_won",
         "_hedges",
         "_now",
+        "_stall_count",
+        "_stall_ms",
         "_started",
         "_terminal_claimed",
         "logical_call_id",
@@ -353,6 +366,8 @@ class _CallContext:
         self._generation_ms = 0
         self._hedges = 0
         self._hedge_won = False
+        self._stall_ms: int | None = None
+        self._stall_count: int | None = None
         self._terminal_claimed = False
 
     def register_attempt(self) -> None:
@@ -363,9 +378,23 @@ class _CallContext:
         """
         self._attempts += 1
 
-    def record_generation(self, elapsed_ms: int, *, accumulate: bool) -> None:
-        """chat/OCR 覆盖(结构化重问最后一轮为准);embedding 分批累加。"""
+    def record_generation(
+        self,
+        elapsed_ms: int,
+        *,
+        accumulate: bool,
+        stall_ms: int | None = None,
+        stall_count: int | None = None,
+    ) -> None:
+        """chat/OCR 覆盖(结构化重问最后一轮为准);embedding 分批累加。
+
+        stall 两值(1.3.8 净口径归因): chat 赢家路传入;embedding/ocr 既有调用
+        恒 None = "无沉默观测"(不是"沉默为 0")——它们是各自链路的墙钟,无中段
+        沉默概念。与 generation 同一次登记覆盖,不单独累加。
+        """
         self._generation_ms = self._generation_ms + elapsed_ms if accumulate else elapsed_ms
+        self._stall_ms = stall_ms
+        self._stall_count = stall_count
 
     def register_hedge(self, *, hedge_won: bool) -> None:
         """对冲路实际发出即计数;赢家裁定后一次性登记。"""
@@ -381,6 +410,8 @@ class _CallContext:
             hedges=self._hedges,
             generation_ms=self._generation_ms,
             hedge_won=self._hedge_won,
+            stall_ms=self._stall_ms,
+            stall_count=self._stall_count,
         )
 
     def claim_terminal(self) -> bool:

tests/unit/test_hedge.py

@@ -569,6 +569,27 @@ class TestHedgeFailureCombination:
         # 两败轮次同样登记对冲路数(设计 §4.5: 实际并发发出即计)
         assert ctx.snapshot().hedges == 2
 
+    async def test_stall_attribution_none_when_no_winner(self):
+        """两败轮次无赢家(1.3.8 排除条款): generation 不记、stall 归因恒 None。
+
+        对冲已实际发出(hedges 照登),但无赢家路 → 裸生成时间与沉默归因都
+        无归属;两路各自的沉默归属各自 attempt 遥测行(那是 T3 的事)。
+        """
+        transport = HedgeTransport(
+            {
+                "s1": [("fail_after", 0.1, lambda: TransientError("p", source_name="s1"))],
+                "s2": [("fail_after", 0.12, lambda: TransientError("h", source_name="s2"))],
+            }
+        )
+        mw, _, _ = _harness([_src("s1"), _src("s2")], transport, max_attempts=1)
+        ctx = _ctx()
+        with pytest.raises(AllSourcesExhausted):
+            await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        stats = ctx.snapshot()
+        assert stats.hedges == 1 and stats.hedge_won is False
+        assert stats.generation_ms == 0
+        assert stats.stall_ms is None and stats.stall_count is None
+
     async def test_both_429_refund_no_budget(self):
         """两路皆 429: 免预算且耗时退 stall 账——小 stall 窗下终局 stalled 而非耗尽。"""
 

tests/unit/test_retry.py

@@ -107,12 +107,15 @@ class _GenClockTransport:
 
     generation_ms 的口径是"只计 transport 调用本身",故推进必须发生在被包
     transport 内部;退避耗时由用例自带的 sleep 闭包推进,与本包装无关。
+    `stalls`(1.3.8 计划 §3.3): 逐次自控上报的 (stall_ms, stall_count),
+    fake 自己报数、不经真实计时;None(默认)= 该次调用无沉默观测。
     """
 
-    def __init__(self, script, advances, clock):
+    def __init__(self, script, advances, clock, stalls=None):
         self._inner = FakeTransport(script)
         self._advances = list(advances)
         self._clock = clock
+        self._stalls = list(stalls) if stalls is not None else None
 
     @property
     def calls(self):
@@ -131,6 +134,7 @@ class _GenClockTransport:
         content_event,
     ):
         advance = self._advances.pop(0)
+        stall = self._stalls.pop(0) if self._stalls is not None else None
         result = await self._inner.complete(
             messages=messages,
             source=source,
@@ -141,6 +145,8 @@ class _GenClockTransport:
             first_token_event=first_token_event,
             content_event=content_event,
         )
+        if stall is not None:
+            result = dataclasses.replace(result, stall_ms=stall[0], stall_count=stall[1])
         self._clock.advance(advance)
         return result
 
@@ -1017,6 +1023,46 @@ class TestGenerationMs:
         stats = ctx.snapshot()
         assert stats.hedges == 0 and stats.hedge_won is False
 
+    async def test_generation_ms_net_of_stall(self):
+        """净口径(1.3.8 设计 §4.2): generation_ms = 墙钟 − Σ(超阈值中段沉默)。
+
+        FakeClock 推进 5.0s 墙钟、fake 自报 stall 3200ms/2 段 → 净 1800ms;
+        归因两值进快照,墙钟还原公式 generation_ms + stall_ms == 5000 成立。
+        推进量取二进制可精确表示值(同类既有用例惯例,防 int 截断抖动)。
+        """
+        clock = FakeClock()
+        transport = _GenClockTransport([_ok()], [5.0], clock, stalls=[(3200.0, 2)])
+        mw, *_ = _harness([_src("a")], [], clock=clock, transport=transport)
+        ctx = self._ctx(clock)
+        resp = await mw(dataclasses.replace(_REQ, call_context=ctx))
+        assert resp.content == "ok"
+        stats = ctx.snapshot()
+        assert stats.generation_ms == 1800
+        assert stats.stall_ms == 3200 and stats.stall_count == 2
+        assert stats.generation_ms + stats.stall_ms == 5000  # 墙钟还原公式
+
+    async def test_generation_ms_clamps_negative_to_zero(self):
+        """沉默上报超过墙钟(异常数据/钟偏差防御) → 净口径 clamp 到 0(N4),归因原样保留。"""
+        clock = FakeClock()
+        transport = _GenClockTransport([_ok()], [5.0], clock, stalls=[(6000.0, 3)])
+        mw, *_ = _harness([_src("a")], [], clock=clock, transport=transport)
+        ctx = self._ctx(clock)
+        await mw(dataclasses.replace(_REQ, call_context=ctx))
+        stats = ctx.snapshot()
+        assert stats.generation_ms == 0
+        assert stats.stall_ms == 6000 and stats.stall_count == 3
+
+    async def test_non_stream_stall_none_no_deduction(self):
+        """stall None(非流式/增量 chunk<2,排除条款) → 净口径不退减(=墙钟),归因 None。"""
+        clock = FakeClock()
+        transport = _GenClockTransport([_ok()], [5.0], clock)  # 不报 stall = 无观测
+        mw, *_ = _harness([_src("a")], [], clock=clock, transport=transport)
+        ctx = self._ctx(clock)
+        await mw(dataclasses.replace(_REQ, call_context=ctx))
+        stats = ctx.snapshot()
+        assert stats.generation_ms == 5000
+        assert stats.stall_ms is None and stats.stall_count is None
+
 
 class TestLogicalAttemptCounting:
     """尝试登记在 transport 调用**之前**(1.3.5 设计 §4)。

tests/unit/test_types.py

@@ -690,6 +690,36 @@ class TestCallStatsAndContext:
         stats = ctx.snapshot()
         assert (stats.hedges, stats.generation_ms, stats.hedge_won) == (1, 200, True)
 
+    def test_callstats_stall_fields_default_none(self):
+        """1.3.8 归因两字段默认 None: 仅旧参数构造不炸(1.3.7 及以前兼容形态)。"""
+        from polygateway.types import CallStats
+
+        stats = CallStats(logical_call_id="lc-1", attempts=2, total_latency_ms=15)
+        assert stats.stall_ms is None
+        assert stats.stall_count is None
+        explicit = CallStats(
+            logical_call_id="lc-2",
+            attempts=2,
+            total_latency_ms=15,
+            stall_ms=3200,
+            stall_count=2,
+        )
+        assert (explicit.stall_ms, explicit.stall_count) == (3200, 2)
+
+    def test_callcontext_record_generation_with_stall_and_snapshot(self):
+        """stall 两值经 record_generation 进 snapshot;不传恒 None(embedding/ocr 形态)。"""
+        from polygateway.types import _CallContext
+
+        ctx = _CallContext(now=_FakeMonotonic())
+        ctx.record_generation(1800, accumulate=False, stall_ms=3200, stall_count=2)
+        stats = ctx.snapshot()
+        assert (stats.generation_ms, stats.stall_ms, stats.stall_count) == (1800, 3200, 2)
+        # 不传(embedding/ocr 既有调用形态): 归因恒 None,None 是"无观测"不是 0
+        plain = _CallContext(now=_FakeMonotonic())
+        plain.record_generation(500, accumulate=False)
+        plain_stats = plain.snapshot()
+        assert plain_stats.stall_ms is None and plain_stats.stall_count is None
+
     def test_context_counts_attempts_and_freezes_elapsed(self):
         """快照是同步冻结的时间切片: 登记两次尝试后耗时按注入钟折算成毫秒。"""
         from polygateway.types import _CallContext