feat: add opt-in mid-stream stall hedging for chat

HEDGE__STALL_AFTER_S triggers a cross-source hedge when the primary
leg stalls mid-stream (independent of HEDGE__AFTER_S); two-phase
hedge window with event recheck discipline; watchdog-killed primary
converges on the surviving hedge leg. STREAM__STALL_THRESHOLD_MS
(1000ms default) adjustable via env/settings/constructor. Assembly
guards: ValueError on never-opening windows, embedding/OCR scope
warning (B4). Batch D 14 tests red-to-green, 1688 unit+contracts.

src/polygateway/client.py

@@ -239,6 +239,8 @@ class GatewayClient:
         call_deadline_s: float | None = None,
         hedge_after_s: float | None = None,
         hedge_max_extra: int = 1,
+        hedge_stall_after_s: float | None = None,
+        stall_threshold_ms: int | None = None,
         now: Any = time.monotonic,
         sleep: Any = asyncio.sleep,
         rng: Any = random.random,
@@ -247,15 +249,25 @@ class GatewayClient:
         self._call_deadline_s = ensure_call_deadline(
             call_deadline_s, "GatewayClient(call_deadline_s=...)"
         )
-        # 对冲守卫与 GatewaySettings 共用同一份(issue #24 H4): 直接构造这条路
-        # 不经过 settings,值域/交叉守卫若只挂在 settings 上就会被它绕过
-        self._hedge_after_s = check_hedge_assembly(
+        # 对冲守卫与 GatewaySettings 共用同一份(issue #24 H4 / 1.3.8 #28): 直接构造这条路
+        # 不经过 settings,值域/交叉守卫若只挂在 settings 上就会被它绕过。
+        # 返回契约(1.3.8 审查 B2): tuple(归一化 after_s, 归一化 stall_after_s)
+        self._hedge_after_s, self._hedge_stall_after_s = check_hedge_assembly(
             hedge_after_s=hedge_after_s,
             hedge_max_extra=hedge_max_extra,
             sources=sources,
             call_deadline_s=self._call_deadline_s,
             origin="GatewayClient(hedge_after_s=...)",
+            stall_after_s=hedge_stall_after_s,
+            stall_origin="GatewayClient(hedge_stall_after_s=...)",
         )
+        # 沉默计量阈值分发(1.3.8 设计 §6): None = 保留各 SourceConfig 自带值
+        # (不 stomp 直构造调用方的 per-source 设置);非 None 经 replace 写进每个源
+        # (replace 触发 SourceConfig.__post_init__ 重跑,值域校验复用同一处)
+        if stall_threshold_ms is not None:
+            sources = [
+                dataclasses.replace(s, stall_threshold_ms=stall_threshold_ms) for s in sources
+            ]
         emitter = (
             TelemetryEmitter(telemetry, scope=scope, pricing=pricing, text_cap=text_cap)
             if telemetry is not None
@@ -280,6 +292,8 @@ class GatewayClient:
             emitter=emitter,
             # max_extra 不下传(issue #24 H5): v1 编排固定单路对冲,无消费者
             hedge_after_s=self._hedge_after_s,
+            # 中段沉默对冲阈值(1.3.8 #28): 独立于 hedge_after_s 生效,守卫归一化值
+            stall_after_s=self._hedge_stall_after_s,
             now=now,
             sleep=sleep,
             rng=rng,
@@ -524,6 +538,9 @@ class GatewayClient:
             call_deadline_s=settings.call_deadline_s,
             hedge_after_s=settings.hedge_after_s,
             hedge_max_extra=settings.hedge_max_extra,
+            hedge_stall_after_s=settings.hedge_stall_after_s,
+            # settings 值为权威,恒分发(1.3.8 §3.5): 含缺省 1000 在内
+            stall_threshold_ms=settings.stall_threshold_ms,
         )
         _mark_owned_components(client, limiter=limiter, breaker=breaker, telemetry=telemetry)
         client._owns_cache = cache is None  # 缓存后端可以是 None(backend=none),helper 会跳过

src/polygateway/config.py

@@ -54,7 +54,7 @@ _SOURCE_FIELDS: dict[str, tuple[str, str]] = {
     "TRUST_ENV": ("trust_env", "bool"),
     "EXTRA_BODY": ("extra_body", "json"),
 }
-_RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE", "HEDGE"})
+_RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE", "HEDGE", "STREAM"})
 _SELECTORS = frozenset({"round_robin", "least_inflight", "health_aware"})
 _QUOTA_FULL = frozenset({"wait", "fail_fast"})
 # 熔断全拒时的处置(issue #14);值域与 _QUOTA_FULL 相同但语义不同——配额满是
@@ -194,6 +194,16 @@ class GatewaySettings:
     # 由 `check_hedge_assembly` 单一定义点把关,直接构造/replace/env 三路一致
     hedge_after_s: float | None = None
     hedge_max_extra: int = 1
+    # 中段沉默对冲(1.3.8 issue #28): 相邻增量 chunk 间隔超阈值 → 并发异源副路。
+    # 缺省 None = 关闭;独立于 hedge_after_s 生效(设计 §9-④);仅对流式 chat 请求
+    # 生效,非流式请求运行期静默退化(入口门)。值域与交叉守卫同归
+    # `check_hedge_assembly` 单一定义点
+    hedge_stall_after_s: float | None = None
+    # 沉默计量阈值(1.3.8 issue #27): 相邻增量 chunk 间隔**严格大于**它才计入
+    # stall_ms 并从 generation_ms 扣减。默认 1000ms(设计 §9-③ 拍板,三层可调);
+    # 只影响计量口径,不改任何行为。值域(正整数)归 `SourceConfig.__post_init__`
+    # 一处——分发经 `dataclasses.replace` 时重跑,本字段不重复校验
+    stall_threshold_ms: int = 1000
 
     def __post_init__(self) -> None:
         self._normalize()
@@ -378,17 +388,17 @@ class GatewaySettings:
         env 路的值域错误已在 `_load_hedge` 里带真实键名报过;此处对合法值是幂等
         空操作,交叉守卫(阈值 vs timeout/deadline/ttft、单源)只在这里有一处。
         """
-        object.__setattr__(
-            self,
-            "hedge_after_s",
-            check_hedge_assembly(
-                hedge_after_s=self.hedge_after_s,
-                hedge_max_extra=self.hedge_max_extra,
-                sources=self.sources,
-                call_deadline_s=self.call_deadline_s,
-                origin="GatewaySettings.hedge_after_s",
-            ),
+        after_s, stall_after_s = check_hedge_assembly(
+            hedge_after_s=self.hedge_after_s,
+            hedge_max_extra=self.hedge_max_extra,
+            sources=self.sources,
+            call_deadline_s=self.call_deadline_s,
+            origin="GatewaySettings.hedge_after_s",
+            stall_after_s=self.hedge_stall_after_s,
+            stall_origin="GatewaySettings.hedge_stall_after_s",
         )
+        object.__setattr__(self, "hedge_after_s", after_s)
+        object.__setattr__(self, "hedge_stall_after_s", stall_after_s)
 
     @classmethod
     def from_env(
@@ -420,6 +430,7 @@ class GatewaySettings:
             circuit_open=_load_choice(env, f"{scope_u}__CIRCUIT_OPEN", _CIRCUIT_OPEN, "fail_fast"),
             call_deadline_s=_load_call_deadline(scope_u, env),
             **_load_hedge(scope_u, env),
+            **_load_stream(scope_u, env),
             **_load_pgw(env),
         )
 
@@ -742,8 +753,29 @@ def _load_call_deadline(scope: str, env: Mapping[str, str]) -> float | None:
     return ensure_call_deadline(_cast(found[1], "float", found[0]), found[0])
 
 
+def _load_stream(scope: str, env: Mapping[str, str]) -> dict[str, object]:
+    """读 `{SCOPE}__STREAM__STALL_THRESHOLD_MS`(1.3.8 issue #27 设计 §6)。
+
+    3 段键(`split("__")` 长度 3 ≠ 4),`_load_sources` 的段数判据天然跳过;
+    `STREAM` 已进 `_RESERVED_SEGMENTS`,4 段的 `{SCOPE}__STREAM__{N}__*` 也不会
+    被当成 provider 段造出源(1.3.7 `HEDGE` 同款)。未设 = 1000(设计 §9-③ 拍板);
+    值域(正整数)归 `SourceConfig.__post_init__` 一处——字段分发经
+    `dataclasses.replace` 时重跑,本函数只负责解析。
+
+    Args:
+        scope: 已大写的 scope 名。
+        env: 已合并的环境映射。
+
+    Returns:
+        `{"stall_threshold_ms": int}`,直传构造器。
+    """
+    found = _first(env, f"{scope}__STREAM__STALL_THRESHOLD_MS")
+    threshold = int(_cast(found[1], "int", found[0])) if found else 1000
+    return {"stall_threshold_ms": threshold}
+
+
 def _load_hedge(scope: str, env: Mapping[str, str]) -> dict[str, object]:
-    """读 `{SCOPE}__HEDGE__AFTER_S`/`{SCOPE}__HEDGE__MAX_EXTRA`(issue #24 H4)。
+    """读 `{SCOPE}__HEDGE__AFTER_S`/`MAX_EXTRA`/`STALL_AFTER_S`(issue #24 H4 + 1.3.8 #28)。
 
     两键均为 3 段键(`split("__")` 长度 3 ≠ 4),`_load_sources` 的段数判据天然
     跳过它们;`HEDGE` 已进 `_RESERVED_SEGMENTS`,4 段的 `{SCOPE}__HEDGE__{N}__*`
@@ -766,7 +798,13 @@ def _load_hedge(scope: str, env: Mapping[str, str]) -> dict[str, object]:
     )
     found_extra = _first(env, f"{scope}__HEDGE__MAX_EXTRA")
     extra = int(_cast(found_extra[1], "int", found_extra[0])) if found_extra else 1
-    return {"hedge_after_s": after, "hedge_max_extra": extra}
+    found_stall = _first(env, f"{scope}__HEDGE__STALL_AFTER_S")
+    stall = (
+        ensure_call_deadline(_cast(found_stall[1], "float", found_stall[0]), found_stall[0])
+        if found_stall
+        else None
+    )
+    return {"hedge_after_s": after, "hedge_max_extra": extra, "hedge_stall_after_s": stall}
 
 
 def check_hedge_assembly(
@@ -776,26 +814,32 @@ def check_hedge_assembly(
     sources: Sequence[SourceConfig],
     call_deadline_s: float | None,
     origin: str,
-) -> float | None:
-    """对冲装配守卫的唯一事实源(issue #24 设计 §5 全表,H4 批准)。
+    stall_after_s: float | None = None,
+    stall_origin: str = "hedge_stall_after_s",
+) -> tuple[float | None, float | None]:
+    """对冲装配守卫的唯一事实源(issue #24 设计 §5 全表,H4 批准 + 1.3.8 issue #28 §6)。
 
     `GatewaySettings.__post_init__` 与 `GatewayClient.__init__` 调同一份,两条装配
-    路的值域/交叉守卫不漂移。返回归一化后的 `hedge_after_s`(None 或有限正数,
-    复用 `ensure_call_deadline` 的值域校验);`hedge_after_s is None`(未启用)时
-    值域归一化后直接返回,交叉守卫不查——它们没有可校验的对象。
+    路的值域/交叉守卫不漂移。返回契约(1.3.8 审查 B2):
+    `tuple(归一化 hedge_after_s, 归一化 stall_after_s)`(各为 None 或有限正数,
+    复用 `ensure_call_deadline` 的值域校验);**两阈值皆 None 才早退**——只对
+    `hedge_after_s is None` 早退会让仅 stall 配置绕过交叉守卫(审查 B2 钉死)。
 
     Args:
-        hedge_after_s: 对冲触发阈值(秒);None = 关闭。
+        hedge_after_s: 首 token/整体耗时对冲触发阈值(秒);None = 关闭。
         hedge_max_extra: 每次逻辑调用最多并发对冲路数;v1 仅单路生效(H5)。
-        sources: 本 scope 的源集合(交叉守卫要读 timeout_s/ttft_timeout_s)。
+        sources: 本 scope 的源集合(交叉守卫要读 timeout_s/ttft/inter_token)。
         call_deadline_s: 调用期限(秒);与对冲的组合守卫见设计 §6。
         origin: after_s 值域报错的定位串(env 键名 / `GatewaySettings.hedge_after_s` /
             `GatewayClient(hedge_after_s=...)`);跨字段守卫与 warning 沿用本模块先例,
             消息自带字段名与 env 键型,不挂 origin。
+        stall_after_s: 中段沉默对冲触发阈值(秒,1.3.8 issue #28);None = 关闭。
+        stall_origin: stall_after_s 值域报错的定位串(同 origin 的三路纪律)。
 
     Raises:
-        ValueError: max_extra 非 int/bool 或出 [1,3];阈值 ≥ 最小源 timeout_s(永不
-            可能触发);阈值 ≥ call_deadline_s(期限先于对冲触发,对冲形同虚设)。
+        ValueError: max_extra 非 int/bool 或出 [1,3];after_s ≥ 最小源 timeout_s(永不
+            可能触发);stall_after_s ≥ 最小源有效 inter-token 超时(回落同款,窗永不开);
+            阈值 ≥ call_deadline_s(期限先于对冲触发,对冲形同虚设)。
     """
     if isinstance(hedge_max_extra, bool) or not isinstance(hedge_max_extra, int):
         raise ValueError(
@@ -807,45 +851,80 @@ def check_hedge_assembly(
             "每次逻辑调用最多并发对冲路数,v1 仅单路生效"
         )
     after = ensure_call_deadline(hedge_after_s, origin)
-    if after is None:
-        return None
+    stall = ensure_call_deadline(stall_after_s, stall_origin)
+    if after is None and stall is None:
+        return None, None
     if not sources:
-        # 直传路(GatewayClient(sources=[], hedge_after_s=...))没有 settings 的
-        # 非空守卫先行拦截;报错必须定位到 hedge,而不是裸 min() 空序列异常
+        # 直传路(GatewayClient(sources=[], ...))没有 settings 的
+        # 非空守卫先行拦截;报错必须定位到 hedge 键族,而不是裸 min() 空序列异常。
+        # 消息保留 "hedge_after_s" 字样: 既有用例据此定位键族(1.3.8 扩含 stall)
         raise ValueError(
-            "hedge_after_s({SCOPE}__HEDGE__AFTER_S)的交叉守卫要求 sources 不能为空;"
-            "对冲已启用但没有可校验的源"
+            "hedge_after_s/hedge_stall_after_s({SCOPE}__HEDGE__AFTER_S/STALL_AFTER_S)"
+            "的交叉守卫要求 sources 不能为空;对冲已启用但没有可校验的源"
         )
     min_timeout = min(s.timeout_s for s in sources)
-    if after >= min_timeout:
-        raise ValueError(
-            f"hedge_after_s({{SCOPE}}__HEDGE__AFTER_S={after})须 < 最小源 timeout_s"
-            f"({min_timeout});对冲永不可能触发,配置即错误"
-        )
-    if call_deadline_s is not None and after >= call_deadline_s:
-        raise ValueError(
-            f"hedge_after_s({after})须 < call_deadline_s({call_deadline_s});"
-            "期限会先于对冲触发,对冲形同虚设(设计 §6)"
-        )
-    ttfts = [s.ttft_timeout_s for s in sources if s.ttft_timeout_s is not None]
-    if ttfts and after >= min(ttfts):
-        logger.warning(
-            "hedge_after_s({}) ≥ 最小源 ttft_timeout_s({}): 流式挂起会被 TTFT 看门狗"
-            "先行切断,对冲对流式形同虚设(非流式仍有效)",
-            after,
-            min(ttfts),
-        )
+    if after is not None:
+        if after >= min_timeout:
+            raise ValueError(
+                f"hedge_after_s({{SCOPE}}__HEDGE__AFTER_S={after})须 < 最小源 timeout_s"
+                f"({min_timeout});对冲永不可能触发,配置即错误"
+            )
+        if call_deadline_s is not None and after >= call_deadline_s:
+            raise ValueError(
+                f"hedge_after_s({after})须 < call_deadline_s({call_deadline_s});"
+                "期限会先于对冲触发,对冲形同虚设(设计 §6)"
+            )
+        ttfts = [s.ttft_timeout_s for s in sources if s.ttft_timeout_s is not None]
+        if ttfts and after >= min(ttfts):
+            logger.warning(
+                "hedge_after_s({}) ≥ 最小源 ttft_timeout_s({}): 流式挂起会被 TTFT 看门狗"
+                "先行切断,对冲对流式形同虚设(非流式仍有效)",
+                after,
+                min(ttfts),
+            )
+    if stall is not None:
+        # 有效最小值与 openai_compat.py:573 看门狗装配的回落同款:
+        # 配了 watchdog 的源按 inter_token_timeout_s,未配的按 timeout_s(设计 §11-N5)
+        min_inter = min((s.inter_token_timeout_s or s.timeout_s) for s in sources)
+        if stall >= min_inter:
+            raise ValueError(
+                f"hedge_stall_after_s({{SCOPE}}__HEDGE__STALL_AFTER_S={stall})须 < 最小源"
+                f"有效 inter-token 超时({min_inter};inter_token_timeout_s 缺省回落"
+                f"timeout_s,与看门狗装配同款);中段对冲窗永不开,配置即错误"
+            )
+        if call_deadline_s is not None and stall >= call_deadline_s:
+            raise ValueError(
+                f"hedge_stall_after_s({stall})须 < call_deadline_s({call_deadline_s});"
+                "期限会先于中段对冲触发,对冲形同虚设(设计 §6)"
+            )
     if len(sources) == 1:
         logger.warning(
-            "单源 scope 配置了对冲阈值 hedge_after_s={};运行期拿不到异源候选,对冲自然静默",
+            "单源 scope 配置了对冲阈值 hedge_after_s={}/hedge_stall_after_s={};"
+            "运行期拿不到异源候选,对冲自然静默",
             after,
+            stall,
         )
     if hedge_max_extra > 1:
         logger.warning(
             "hedge_max_extra={} 已接受,但 v1 仅单路对冲生效(梯次追加为 H5 预留)",
             hedge_max_extra,
         )
-    return after
+    return after, stall
+
+
+def _warn_hedge_keys_on_non_stream_scope(gateway: GatewaySettings, scope_kind: str) -> None:
+    """B4(1.3.8 设计 §11): 装配期可证明纯非流式的 scope 配了对冲键 → warning。
+
+    chat scope 不做(`stream` 是 per-call 参数,装配期无信号,B4 裁决);对冲键在
+    Embedding/OCR 链路没有任何消费者,静默退化必须说出来。`hedge_after_s` 一并
+    检测(1.3.7 同病顺带补齐,缺陷修复不扩公共面)。
+    """
+    if gateway.hedge_after_s is not None or gateway.hedge_stall_after_s is not None:
+        logger.warning(
+            "对冲键(HEDGE__AFTER_S/HEDGE__STALL_AFTER_S)仅对 chat scope 生效: "
+            "{} scope 非流式,对冲配置静默退化(1.3.8 B4)",
+            scope_kind,
+        )
 
 
 @dataclass(frozen=True)
@@ -868,6 +947,7 @@ class EmbeddingSettings:
             raise ValueError(f"EmbeddingSettings.batch_size 必须 ≥ 1: {self.batch_size}")
         if self.expected_dim is not None and self.expected_dim < 1:
             raise ValueError(f"EmbeddingSettings.expected_dim 必须 ≥ 1: {self.expected_dim}")
+        _warn_hedge_keys_on_non_stream_scope(self.gateway, "embedding")
 
     @classmethod
     def from_env(
@@ -911,6 +991,10 @@ class OcrSettings:
 
     gateway: GatewaySettings
 
+    def __post_init__(self) -> None:
+        """内嵌 gateway 由 `GatewaySettings.__post_init__` 自己把关;此处只补 B4 对冲键 warning。"""
+        _warn_hedge_keys_on_non_stream_scope(self.gateway, "ocr")
+
     @classmethod
     def from_env(
         cls,

src/polygateway/middleware/retry.py

@@ -201,6 +201,7 @@ class RetryMW:
         pacer: AdaptivePacer | None = None,
         emitter: TelemetryEmitter | None = None,
         hedge_after_s: float | None = None,
+        stall_after_s: float | None = None,
         now: Callable[[], float] = time.monotonic,
         sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
         rng: Callable[[], float] = random.random,
@@ -222,6 +223,9 @@ class RetryMW:
         # 值域/交叉守卫在装配层(config.check_hedge_assembly),本类不重复校验;
         # hedge_max_extra 不下传——v1 编排固定单路对冲(H5)
         self._hedge_after_s = hedge_after_s
+        # 中段沉默对冲触发阈值(1.3.8 issue #28): None = 关闭;独立于 _hedge_after_s
+        # 生效(设计 §9-④),仅对流式请求开窗。守卫同在装配层,本类不重复校验
+        self._stall_after_s = stall_after_s
         self._now = now
         self._sleep = sleep
         self._rng = rng
@@ -272,8 +276,20 @@ class RetryMW:
                 # (单路 = 成功那轮;对冲 = 赢家那一路),`_attempt` 只负责把本次
                 # transport 耗时与沉默观测投进来
                 generation_sink: list[_GenerationSample] = []
-                if self._hedge_after_s is None:
-                    # 默认关闭: 逐字 1.3.6 单路路径(默认关闭回归门据此成立)
+                if self._hedge_after_s is None and self._stall_after_s is None:
+                    # 两阈值均未设: 逐字 1.3.7 单路路径(默认关闭回归门据此成立)
+                    outcome = await self._attempt(
+                        request,
+                        *picked,
+                        reasons,
+                        attempt_fails,
+                        first_token_event=None,
+                        content_event=None,
+                        generation_sink=generation_sink,
+                    )
+                elif self._hedge_after_s is None and not request.stream:
+                    # 仅 stall 已设且请求非流式: 中段沉默无非流式概念,逐字走单路
+                    # (等价未配置,设计 §11-B3 入口门)
                     outcome = await self._attempt(
                         request,
                         *picked,
@@ -374,7 +390,15 @@ class RetryMW:
             self._feed_outcome(source.name, ok=True)
             self._pacer.on_success(source.name)
             response = self._build_response(source, result, call_id, started)
-            await self._emit(request, source, call_id, started, response=response)
+            await self._emit(
+                request,
+                source,
+                call_id,
+                started,
+                response=response,
+                stall_ms=stall_ms_i,
+                stall_count=stall_count_i,
+            )
             return response
         except RequestRejectedError as exc:
             actual, settlement_known = 0, True  # 逐字保住 1.3.5: 坏请求全额退还
@@ -441,6 +465,11 @@ class RetryMW:
         # Phase 1 启动原路: 事件与 sink 每轮新建(局部状态,严禁实例属性)
         source, permit, entry = picked
         first_token: asyncio.Event = asyncio.Event()
+        # 中段沉默观测(1.3.8): 仅"流式 + stall 阈值已设"的原路携带真事件;
+        # 其余恒 None = 不观测(对冲路亦恒 None,Phase 4,N3③)
+        content_event = (
+            asyncio.Event() if (request.stream and self._stall_after_s is not None) else None
+        )
         sink_p: list[_GenerationSample] = []
         primary = asyncio.create_task(
             self._attempt(
@@ -451,14 +480,15 @@ class RetryMW:
                 reasons,
                 attempt_fails,
                 first_token_event=first_token,
-                content_event=None,
+                content_event=content_event,
                 generation_sink=sink_p,
             )
         )
         hedge: asyncio.Task[LLMResponse | _Failed] | None = None
         try:
-            # Phase 2 触发窗: 只认"阈值到 + 首 token 未至 + 原路在途"(loop 相对时长)
-            if not await self._past_hedge_window(primary, first_token):
+            # Phase 2 触发窗: 两阶段(1.3.8)——首 token 前走 AFTER_S,首 token 后
+            # 走 STALL_AFTER_S;只认 loop 相对时长,绝不读注入钟(设计 §4.1)
+            if not await self._past_hedge_window(primary, first_token, content_event):
                 # 原路已了结/首 token 已至: 等价于未配置对冲
                 outcome = await primary
                 if isinstance(outcome, LLMResponse):
@@ -558,19 +588,32 @@ class RetryMW:
                 await asyncio.gather(*tasks, return_exceptions=True)
             raise
 
-    async def _past_hedge_window(self, primary: asyncio.Task, first_token: asyncio.Event) -> bool:
-        """对冲触发窗: 阈值到 + 首 token 未至 + 原路在途,三者齐备才放行对冲。
+    async def _past_hedge_window(
+        self,
+        primary: asyncio.Task,
+        first_token: asyncio.Event,
+        content_event: asyncio.Event | None,
+    ) -> bool:
+        """对冲触发窗两阶段(1.3.8 设计 §5.2): 首 token 前走 AFTER_S,首 token 后走 STALL_AFTER_S。
 
         只用事件循环相对时长(`asyncio.wait` timeout),绝不读注入 `now`——测试
         伪造注入钟跳变不得触发对冲(设计 §4.1)。waiter 任务必须收口,且**不得吞
         外部取消**(finally 里 await 已取消的 waiter 会接住一个 CancelledError,
         须靠 cancelling() 区分它是 waiter 自己的还是外面打进来的)。
         """
+        # 阶段 1 首 token 前: AFTER_S 已设才开窗(设计 §5.2);仅 stall 时此阶段不
+        # 触发,但仍须等首 token 到达才能进阶段 2——否则 pre-TTFT 沉默会被 stall
+        # 窗误吞(计划 §3.6)
         waiter = asyncio.create_task(first_token.wait())
         try:
-            await asyncio.wait(
-                {primary, waiter}, timeout=self._hedge_after_s, return_when=asyncio.FIRST_COMPLETED
-            )
+            if self._hedge_after_s is not None:
+                await asyncio.wait(
+                    {primary, waiter},
+                    timeout=self._hedge_after_s,
+                    return_when=asyncio.FIRST_COMPLETED,
+                )
+            else:
+                await asyncio.wait({primary, waiter}, return_when=asyncio.FIRST_COMPLETED)
         finally:
             waiter.cancel()
             try:
@@ -578,7 +621,35 @@ class RetryMW:
             except asyncio.CancelledError:
                 if asyncio.current_task().cancelling():  # 外部取消,穿透
                     raise
-        return not primary.done() and not first_token.is_set()
+        if self._hedge_after_s is not None and not primary.done() and not first_token.is_set():
+            return True  # AFTER_S 触发: 阈值到 + 首 token 未至 + 原路在途(1.3.7 原语义)
+        if primary.done():
+            return False
+        # 阶段 2 首 token 后: STALL_AFTER_S 循环等待;非流式/未设 stall → 不开窗
+        if self._stall_after_s is None or content_event is None:
+            return False
+        while not primary.done():
+            waiter = asyncio.create_task(content_event.wait())  # 每轮自建 waiter(N3④)
+            try:
+                await asyncio.wait(
+                    {primary, waiter},
+                    timeout=self._stall_after_s,
+                    return_when=asyncio.FIRST_COMPLETED,
+                )
+            finally:
+                waiter.cancel()
+                try:
+                    await waiter
+                except asyncio.CancelledError:
+                    if asyncio.current_task().cancelling():  # 外部取消,穿透
+                        raise
+            if primary.done():
+                return False
+            if content_event.is_set():  # N3① 复检: chunk 与超时同 pass 到达 → 防误触发
+                content_event.clear()  # N3② check→clear 之间无 await
+                continue
+            return True  # 复检仍未置位 = 真沉默 → 触发对冲
+        return False
 
     @staticmethod
     def _succeeded(task: asyncio.Task, done: set[asyncio.Task]) -> bool:
@@ -700,11 +771,15 @@ class RetryMW:
         *,
         response: LLMResponse | None = None,
         error: PolyGatewayError | str | None = None,
+        stall_ms: int | None = None,
+        stall_count: int | None = None,
     ) -> None:
         """逐次遥测(经注入的单一 Emitter);遥测失败不得影响调用(铁律)。
 
         异常**对象原样下传**而非先 `str()` 压平(1.3.5 设计 §5): 状态码、底层异常
         类型与网关响应体已经在异常上了,在这里压平就是把它们丢掉。
+        stall 两值(1.3.8)只在成功分支由 `_attempt` 显式传入(计划 §3.4);
+        失败/取消分支恒 None——被取消输家无 TransportResult(设计 §11-B2)。
         """
         if self._emitter is None:
             return
@@ -720,6 +795,8 @@ class RetryMW:
                 reasoning_applies=True,
                 # 公开方法四值之一;本中间件只服务 chat 洋葱
                 operation="chat",
+                stall_ms=stall_ms,
+                stall_count=stall_count,
             )
         except asyncio.CancelledError:
             raise