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(
+        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,16 +851,19 @@ 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 is not None:
         if after >= min_timeout:
             raise ValueError(
                 f"hedge_after_s({{SCOPE}}__HEDGE__AFTER_S={after})须 < 最小源 timeout_s"
@@ -835,17 +882,49 @@ def check_hedge_assembly(
                 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,18 +588,53 @@ 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:
+            if self._hedge_after_s is not None:
                 await asyncio.wait(
-                {primary, waiter}, timeout=self._hedge_after_s, return_when=asyncio.FIRST_COMPLETED
+                    {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:
+                await waiter
+            except asyncio.CancelledError:
+                if asyncio.current_task().cancelling():  # 外部取消,穿透
+                    raise
+        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()
@@ -578,7 +643,13 @@ class RetryMW:
                 except asyncio.CancelledError:
                     if asyncio.current_task().cancelling():  # 外部取消,穿透
                         raise
-        return not primary.done() and not first_token.is_set()
+            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

tests/unit/test_config.py

@@ -1179,17 +1179,16 @@ class TestHedgeConfig:
                 origin="GatewayClient(hedge_after_s=8)",
             )
         assert "hedge_after_s" in str(ei.value)  # 定位得到是哪个键
-        # 未启用对冲(None)时空 sources 直接放行: 交叉守卫没有可校验的对象
-        assert (
-            check_hedge_assembly(
+        # 未启用对冲(两阈值皆 None)时空 sources 直接放行: 交叉守卫没有可校验的对象;
+        # 返回契约(1.3.8 审查 B2): tuple(归一化 after_s, 归一化 stall_after_s),
+        # 两阈值皆 None 才早退
+        assert check_hedge_assembly(
             hedge_after_s=None,
             hedge_max_extra=1,
             sources=[],
             call_deadline_s=None,
             origin="test",
-            )
-            is None
-        )
+        ) == (None, None)
 
     def test_hedge_guard_below_min_timeout_raises(self):
         """阈值 ≥ 最小源 timeout_s = 对冲永不可能触发,装配期炸掉(ValueError)。"""
@@ -1283,3 +1282,172 @@ class TestHedgeConfig:
         client = GatewayClient.from_settings(s)
         assert client._hedge_after_s == 8.0
         assert client._terminal._hedge_after_s == 8.0
+
+
+class TestStallConfig:
+    """1.3.8 两新键: `{SCOPE}__STREAM__STALL_THRESHOLD_MS`(计量,默认 1000 三层可调)
+    与 `{SCOPE}__HEDGE__STALL_AFTER_S`(中段对冲,未设=关闭)及其装配守卫(设计 §6,
+    计划批次 D;单一定义点仍是 `config.check_hedge_assembly`)。"""
+
+    def _two_source_env(self, **overrides):
+        """双源 env(同 provider 避免注册表依赖): 隔离单源 warning 的干扰。"""
+        return _env(
+            **{
+                "LLM__QWEN__2__BASE_URL": "https://gw-b.example/v1",
+                "LLM__QWEN__2__API_KEY": "sk-b",
+                "LLM__QWEN__2__MODEL": "qwen-plus",
+                "LLM__QWEN__2__TIMEOUT_S": "90",
+                **overrides,
+            }
+        )
+
+    def test_stall_threshold_key_loads_three_layers(self):
+        """计量阈值三层调整接口(§9-③ 人类要求): env 键 / settings 字段 / client kwarg。
+
+        分发语义: from_settings 恒以 settings 值为权威(覆盖源自带值);client 直传
+        None = 保留各 SourceConfig 自带值,非 None 经 dataclasses.replace 分发。"""
+        from tests.unit.test_client import _client
+
+        # 层 1: env 键 → settings 字段 → from_settings 分发进每个 SourceConfig
+        env = self._two_source_env(**{"LLM__STREAM__STALL_THRESHOLD_MS": "2500"})
+        s = GatewaySettings.from_env("LLM", env=env)
+        assert s.stall_threshold_ms == 2500
+        client = GatewayClient.from_settings(s)
+        assert {src.stall_threshold_ms for src in client._terminal._sources} == {2500}
+        # 层 2: settings 字段直接构造(dataclasses.replace 同路),settings 值权威:
+        # 源自带 500 也被 scope 值覆盖(恒分发,计划 §3.5)
+        base = GatewaySettings.from_env("LLM", env=_env())
+        custom_source = dataclasses.replace(base.sources[0], stall_threshold_ms=500)
+        s2 = dataclasses.replace(base, sources=(custom_source,), stall_threshold_ms=800)
+        client2 = GatewayClient.from_settings(s2)
+        assert client2._terminal._sources[0].stall_threshold_ms == 800
+        # 层 3: client kwarg——非 None 分发,None(缺省)保留源自带值
+        client3 = _client(stall_threshold_ms=600)
+        assert client3._terminal._sources[0].stall_threshold_ms == 600
+        client4 = _client(sources=[custom_source])
+        assert client4._terminal._sources[0].stall_threshold_ms == 500  # 不 stomp
+        # 未设恒 1000(SourceConfig 缺省)
+        assert GatewaySettings.from_env("LLM", env=_env()).stall_threshold_ms == 1000
+        assert _client()._terminal._sources[0].stall_threshold_ms == 1000
+
+    def test_stream_segment_reserved(self):
+        """`STREAM` 进保留段: 四段 `LLM__STREAM__1__*` 不造源(1.3.7 HEDGE 同款)。"""
+        env_collision = _env(
+            **{
+                "LLM__STREAM__1__BASE_URL": "https://gw-c.example/v1",
+                "LLM__STREAM__1__API_KEY": "sk-c",
+                "LLM__STREAM__1__MODEL": "m-c",
+                "LLM__STREAM__1__TIMEOUT_S": "60",
+            }
+        )
+        s = GatewaySettings.from_env("LLM", env=env_collision)
+        assert [src.name for src in s.sources] == ["qwen_1"]
+        # 三段键天然跳过 `_load_sources` 的段数判据,也不造源
+        env = _env(**{"LLM__STREAM__STALL_THRESHOLD_MS": "2000"})
+        s2 = GatewaySettings.from_env("LLM", env=env)
+        assert [src.name for src in s2.sources] == ["qwen_1"]
+
+    def test_stall_after_s_guard_effective_minimum_raises(self):
+        """`stall_after_s ≥ min(inter_token_timeout_s or timeout_s)` → ValueError。
+
+        有效值回落与 openai_compat.py:573 看门狗装配同款: 配 watchdog 的源用
+        inter_token_timeout_s,未配的源用 timeout_s;对冲窗永不开装配期直接拒绝。"""
+        # 源 1 配 watchdog(inter_token 15)、源 2 纯 timeout(90): 有效最小值 = 15
+        env = self._two_source_env(
+            **{
+                "LLM__QWEN__1__TTFT_TIMEOUT_S": "30",
+                "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15",
+                "LLM__HEDGE__STALL_AFTER_S": "15",  # ≥ 15 → 拒绝(边界含等号)
+            }
+        )
+        with pytest.raises(ValueError, match="STALL_AFTER_S"):
+            GatewaySettings.from_env("LLM", env=env)
+        s = GatewaySettings.from_env(
+            "LLM",
+            env=self._two_source_env(
+                **{
+                    "LLM__QWEN__1__TTFT_TIMEOUT_S": "30",
+                    "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15",
+                    "LLM__HEDGE__STALL_AFTER_S": "14",
+                }
+            ),
+        )
+        assert s.hedge_stall_after_s == 14.0
+        # 全源未配 watchdog: 有效最小值 = min(timeout_s) = 90
+        with pytest.raises(ValueError, match="STALL_AFTER_S"):
+            GatewaySettings.from_env(
+                "LLM", env=self._two_source_env(**{"LLM__HEDGE__STALL_AFTER_S": "90"})
+            )
+
+    def test_stall_after_s_guard_deadline_conflict_raises(self):
+        """`stall_after_s ≥ call_deadline_s` → ValueError(与 after_s 同款,§11-N5)。"""
+        env = self._two_source_env(
+            **{"LLM__HEDGE__STALL_AFTER_S": "30", "LLM__CALL_DEADLINE_S": "30"}
+        )
+        with pytest.raises(ValueError, match="call_deadline_s"):
+            GatewaySettings.from_env("LLM", env=env)
+        s = GatewaySettings.from_env(
+            "LLM",
+            env=self._two_source_env(
+                **{"LLM__HEDGE__STALL_AFTER_S": "29", "LLM__CALL_DEADLINE_S": "30"}
+            ),
+        )
+        assert s.hedge_stall_after_s == 29.0 and s.call_deadline_s == 30.0
+
+    def test_stall_after_s_single_source_warns(self):
+        """单源 scope 设 stall 阈值: 装配期 warning 放行(1.3.7 §5 同款)。"""
+        with _captured_warnings() as warnings:
+            s = GatewaySettings.from_env(
+                "LLM", env=_env(**{"LLM__HEDGE__STALL_AFTER_S": "8"})
+            )
+        assert s.hedge_stall_after_s == 8.0
+        assert any("单源" in m for m in warnings)
+
+    def test_embedding_ocr_scope_hedge_keys_warn(self):
+        """B4: Embedding/OCR scope 配对冲键(AFTER_S 或 STALL_AFTER_S)→ 装配 warning;
+        chat scope 无此 warning(per-call stream,装配期无信号,B4 裁决)。"""
+        # Embedding: env 路(AFTER_S 与 STALL_AFTER_S 各验一次)
+        embed_env = {
+            "EMBED__QWEN__1__BASE_URL": "https://gw.example/v1",
+            "EMBED__QWEN__1__API_KEY": "sk-a",
+            "EMBED__QWEN__1__MODEL": "text-embedding-v3",
+            "EMBED__QWEN__1__TIMEOUT_S": "60",
+            "EMBED__RETRY__MAX_ATTEMPTS": "3",
+            "EMBED__RETRY__BACKOFF_BASE_S": "1.0",
+            "EMBED__RETRY__BACKOFF_MAX_S": "10.0",
+            "EMBED__BREAKER__FAIL_THRESHOLD": "5",
+            "EMBED__BREAKER__COOLDOWN_S": "60",
+            "PGW_CACHE_BACKEND": "none",
+            "PGW_TELEMETRY_BACKEND": "none",
+            "EMBED__BATCH_SIZE": "64",
+        }
+        with _captured_warnings() as warnings:
+            EmbeddingSettings.from_env(
+                "EMBED", env={**embed_env, "EMBED__HEDGE__AFTER_S": "8"}
+            )
+        assert any("仅对 chat scope 生效" in m for m in warnings)
+        with _captured_warnings() as warnings:
+            EmbeddingSettings.from_env(
+                "EMBED", env={**embed_env, "EMBED__HEDGE__STALL_AFTER_S": "8"}
+            )
+        assert any("仅对 chat scope 生效" in m for m in warnings)
+        # Embedding: 直构造路(__post_init__ 那道,非 from_env 专属)
+        gw = GatewaySettings.from_env("EMBED", env=embed_env)
+        gw_hedged = dataclasses.replace(gw, hedge_stall_after_s=8.0)
+        with _captured_warnings() as warnings:
+            EmbeddingSettings(gateway=gw_hedged, batch_size=64)
+        assert any("仅对 chat scope 生效" in m for m in warnings)
+        # OCR: env 路(1.3.7 AFTER_S 同病顺带补齐 + 新 STALL_AFTER_S)
+        ocr_env = dict(TestOcrSettings._OCR_ENV)
+        with _captured_warnings() as warnings:
+            OcrSettings.from_env("OCR", env={**ocr_env, "OCR__HEDGE__AFTER_S": "8"})
+        assert any("仅对 chat scope 生效" in m for m in warnings)
+        with _captured_warnings() as warnings:
+            OcrSettings.from_env("OCR", env={**ocr_env, "OCR__HEDGE__STALL_AFTER_S": "8"})
+        assert any("仅对 chat scope 生效" in m for m in warnings)
+        # chat 侧: 同样的对冲键不产生 B4 warning(双源 env 隔离单源 warning 干扰)
+        with _captured_warnings() as warnings:
+            GatewaySettings.from_env(
+                "LLM", env=self._two_source_env(**{"LLM__HEDGE__STALL_AFTER_S": "8"})
+            )
+        assert not any("仅对 chat scope 生效" in m for m in warnings)

tests/unit/test_stall_hedge.py

@@ -0,0 +1,479 @@
+"""中段沉默对冲编排测试(1.3.8 issue #28 设计 §5.2 / 计划批次 D)。
+
+设施纪律(计划 §5): 复用 test_hedge.py 范式——两源 scope、事件驱动假 transport、
+**真实 loop 钟**(不注入 FakeClock 驱动触发;注入钟仅用于赢家净口径精确断言,
+触发路径结构性不读它)、阈值 0.05s、断言容差 4–10×;取消窗口用 entered Event
+栅栏钉死,禁 sleep 撞窗口;不标 slow。
+
+命名区隔(审查 N5): 本文件的 `stall_after_s` 是**对冲触发阈值**(相邻增量 chunk
+间隔超过它 → 触发对冲);test_hedge.py:150 的 `stall_window_s` 是 **StallClock
+预算**(非生产性等待的调用级上限)——同名词不同概念,勿混。
+"""
+
+import asyncio
+import dataclasses
+import time
+
+import pytest
+
+from polygateway.backends.memory.breaker import InMemoryGate
+from polygateway.backends.memory.limiter import InMemoryLimiter
+from polygateway.errors import AllSourcesExhausted, TransientError
+from polygateway.middleware.retry import RetryMW
+from polygateway.sources import SourceCooldownMemo
+from polygateway.types import (
+    BackpressurePolicy,
+    BreakerConfig,
+    ChatRequest,
+    GlobalLimits,
+    RetryPolicy,
+    _CallContext,
+)
+from tests.contracts.conftest import FakeClock
+from tests.unit.test_hedge import RecordingEmitter
+from tests.unit.test_retry import StaticSelector, _ok, _src
+
+_BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
+_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
+_STALL_AFTER_S = 0.05  # 真实 loop 钟对冲触发阈值(≠ StallClock 预算 stall_window_s,N5)
+
+
+class StallHedgeTransport:
+    """事件驱动假 transport(中段沉默剧本): 逐步骤控制 chunk 事件/挂起/成败。
+
+    剧本 = 每源一串**步骤**,每次 `complete` 调用从上次位置继续执行到终态步骤
+    (`ok`/`fail`)为止;多轮调用(重试)顺序消费后续步骤,耗尽即 AssertionError。
+
+    步骤:
+      ("chunk",)                      — 置 first_token_event 与 content_event(首个增量 chunk)
+      ("chunk_after", delay)          — 睡 delay 后只置 content_event(后续增量 chunk)
+      ("hang",)                       — 置位该源 entered,挂起直到该源 release(或被取消)
+      ("ok", content)                 — 终态: 立即成功(带已登记的 stall/clock 效果)
+      ("ok_after", delay, content)    — 终态: 真实 loop 钟睡 delay 后成功
+      ("fail_after", delay, factory)  — 终态: 睡 delay 后抛 `factory()` 新造的异常
+      ("stall", ms, count)            — 非终态: 给本剧本终态结果打自报 stall 值(假
+                                        transport 自己报数,不经真实计时,计划 §3.3)
+      ("advance", secs)               — 非终态: 推进注入钟(仅配 FakeClock 的用例)
+    """
+
+    def __init__(self, scripts: dict[str, list[tuple]], clock: FakeClock | None = None):
+        self._scripts = {name: list(steps) for name, steps in scripts.items()}
+        self._pos = dict.fromkeys(scripts, 0)
+        self._clock = clock
+        self.calls: list[str] = []
+        # 逐次记录收到的 first_token_event/content_event 身份(对冲路恒 None 断言原料)
+        self.ft_events: list[object] = []
+        self.ce_events: list[object] = []
+        self.entered = {name: asyncio.Event() for name in scripts}
+        self.release = {name: asyncio.Event() for name in scripts}
+
+    async def complete(
+        self,
+        *,
+        messages,
+        source,
+        stream,
+        overlay,
+        call_id,
+        reasoning_effort,
+        first_token_event,
+        content_event,
+    ):
+        self.calls.append(source.name)
+        self.ft_events.append(first_token_event)
+        self.ce_events.append(content_event)
+        steps = self._scripts[source.name]
+        stall: tuple[float, float] | None = None
+        while self._pos[source.name] < len(steps):
+            action = steps[self._pos[source.name]]
+            self._pos[source.name] += 1
+            kind = action[0]
+            if kind == "chunk":
+                if first_token_event is not None:
+                    first_token_event.set()
+                if content_event is not None:
+                    content_event.set()
+            elif kind == "chunk_after":
+                await asyncio.sleep(action[1])
+                if content_event is not None:
+                    content_event.set()
+            elif kind == "hang":
+                self.entered[source.name].set()
+                await self.release[source.name].wait()
+            elif kind == "stall":
+                stall = (action[1], action[2])
+            elif kind == "advance":
+                assert self._clock is not None, "advance 步骤需要构造时注入 FakeClock"
+                self._clock.advance(action[1])
+            elif kind == "ok":
+                result = _ok(action[1])
+                if stall is not None:
+                    result = dataclasses.replace(result, stall_ms=stall[0], stall_count=stall[1])
+                return result
+            elif kind == "ok_after":
+                await asyncio.sleep(action[1])
+                result = _ok(action[2])
+                if stall is not None:
+                    result = dataclasses.replace(result, stall_ms=stall[0], stall_count=stall[1])
+                return result
+            elif kind == "fail_after":
+                await asyncio.sleep(action[1])
+                raise action[2]()
+            else:
+                raise AssertionError(f"未知剧本步骤: {action!r}")
+        raise AssertionError(f"{source.name} 剧本耗尽且无终态步骤")
+
+
+def _harness(
+    sources,
+    transport,
+    *,
+    hedge_after_s=None,
+    stall_after_s=None,
+    max_attempts=3,
+    emitter=None,
+    selector=None,
+    limiter=None,
+    gate=None,
+    stall_window_s=300.0,
+    now=None,
+):
+    """真实 loop 钟装配(同 test_hedge.py 范式;`now` 仅供净口径精确断言注入 FakeClock)。
+
+    `stall_window_s` 是 StallClock 预算(N5 命名区隔),与 `stall_after_s` 无关。
+    """
+    limiter = limiter or InMemoryLimiter(
+        scope="llm",
+        sources={s.name: s for s in sources},
+        global_limits=_NO_GLOBAL,
+        lease_ttl_s=100.0,
+    )
+    gate = gate or InMemoryGate(config=_BREAKER)
+    mw = RetryMW(
+        scope="llm",
+        sources=sources,
+        # 固定配置序: 原路恒为 s1、对冲路恒为 s2,断言不依赖选源器内部状态
+        selector=selector if selector is not None else StaticSelector(),
+        limiter=limiter,
+        gate=gate,
+        transport=transport,
+        retry=RetryPolicy(max_attempts=max_attempts, backoff_base_s=0.01, backoff_max_s=0.05),
+        backpressure=BackpressurePolicy(stall_window_s=stall_window_s, poll_interval_s=0.01),
+        quota_full="wait",
+        cooldown_memo=SourceCooldownMemo(),
+        emitter=emitter,
+        hedge_after_s=hedge_after_s,
+        stall_after_s=stall_after_s,
+        **({"now": now} if now is not None else {}),
+    )
+    return mw, limiter, gate
+
+
+def _req(*, stream=True, ctx=None):
+    """中段沉默是流式概念,本文件默认 stream=True(test_hedge.py 默认 False 相反)。"""
+    return ChatRequest(
+        messages=[{"role": "user", "content": "hi"}], stream=stream, call_context=ctx
+    )
+
+
+def _ctx(now=None):
+    return _CallContext(now=now if now is not None else time.monotonic)
+
+
+class TestStallHedgeTrigger:
+    """阶段 2 触发(设计 §5.2): 首 token 已至后相邻增量 chunk 间隔超阈值 → 对冲。"""
+
+    async def test_stall_triggers_hedge_after_first_token(self):
+        """s1 首 chunk 后沉默、s2 即时成功: stall 窗触发对冲,赢家为对冲路。
+
+        净口径顺延断言(计划批次 D 首行): 注入 FakeClock + s2 自报 stall
+        (1500ms/2 段)、advance 2.0s → 赢家 `generation_ms == 500` 精确值,
+        `generation_ms + stall_ms == total_latency_ms`(同钟还原公式)。
+        遥测行(T3 缺口补齐钉): 赢家行 stall 两值为自报值,输家行恒 None。
+        """
+        clock = FakeClock()
+        emitter = RecordingEmitter()
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {
+                "s1": [("chunk",), ("hang",), ("ok", "late")],
+                "s2": [("stall", 1500.0, 2), ("advance", 2.0), ("ok", "fast")],
+            },
+            clock=clock,
+        )
+        mw, _, _ = _harness(
+            [s1, s2], transport, stall_after_s=_STALL_AFTER_S, emitter=emitter, now=clock
+        )
+        ctx = _ctx(now=clock)
+        started = time.monotonic()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        elapsed = time.monotonic() - started
+        assert resp.content == "fast" and resp.source_name == "s2"
+        assert elapsed < 10 * _STALL_AFTER_S  # 沉默路被对冲截断,而非等到看门狗
+        assert transport.calls == ["s1", "s2"]
+        stats = ctx.snapshot()
+        assert stats.hedges == 1 and stats.hedge_won is True and stats.attempts == 2
+        # 赢家路净口径: 墙钟 2000(advance 精确推进)− 沉默 1500 = 500
+        assert stats.generation_ms == 500
+        assert stats.stall_ms == 1500 and stats.stall_count == 2
+        assert stats.generation_ms + stats.stall_ms == stats.total_latency_ms  # 还原公式
+        # 逐次遥测: 赢家行带 stall 归因,被取消输家行恒 None(设计 §11-B2)
+        winner = next(r for r in emitter.rows if r["source"] == "s2")
+        loser = next(r for r in emitter.rows if r["source"] == "s1")
+        assert winner["stall_ms"] == 1500 and winner["stall_count"] == 2
+        assert winner["error"] is None
+        assert loser["error"] == "hedge_cancelled"
+        assert loser["stall_ms"] is None and loser["stall_count"] is None
+
+    async def test_single_leg_attempt_row_carries_stall(self):
+        """T3 缺口补齐钉: 未启用对冲的 chat 成功行也须经 `_emit` 落 stall 两值。
+
+        计划 §3.4 指定 `retry._emit` 显式转发(归 T3);批次 C 落行断言只覆盖了
+        emitter 直调,chat 路径漏传无人把守——本用例钉住端到端路径。
+        """
+        clock = FakeClock()
+        emitter = RecordingEmitter()
+        transport = StallHedgeTransport(
+            {"s1": [("stall", 300.0, 1), ("advance", 1.0), ("ok", "x")]}, clock=clock
+        )
+        mw, _, _ = _harness([_src("s1")], transport, emitter=emitter, now=clock)
+        ctx = _ctx(now=clock)
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.content == "x"
+        assert len(emitter.rows) == 1
+        row = emitter.rows[0]
+        assert row["error"] is None
+        assert row["stall_ms"] == 300 and row["stall_count"] == 1
+        stats = ctx.snapshot()
+        assert stats.generation_ms == 700 and stats.stall_ms == 300
+
+    async def test_stall_recheck_prevents_false_trigger(self):
+        """chunk 在超时到期前一刻到达(0.049s 置位,阈值 0.05s)→ 不触发(N3① 复检)。
+
+        只断"不触发 + attempts==1"结果,不断精确时序(计划 §7 抖动纪律)。
+        """
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {
+                "s1": [("chunk",), ("chunk_after", 0.049), ("ok", "done")],
+                "s2": [("ok", "other")],
+            }
+        )
+        mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S)
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.content == "done" and resp.source_name == "s1"
+        assert transport.calls == ["s1"]  # 对冲从未发出
+        stats = ctx.snapshot()
+        assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False
+
+    async def test_stall_and_after_s_independent_matrix(self):
+        """参数化四格(设计 §5.2 逐字): 两阈值独立生效,各格触发语义不串。"""
+        # 格 1: 仅 AFTER_S + 流式 pre-TTFT 挂起 → AFTER_S 触发(1.3.7 语义不变)
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {"s1": [("hang",), ("ok", "late")], "s2": [("ok", "fast")]}
+        )
+        mw, _, _ = _harness([s1, s2], transport, hedge_after_s=_STALL_AFTER_S)
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.source_name == "s2"
+        assert ctx.snapshot().hedges == 1
+
+        # 格 2: 仅 STALL_AFTER_S + 流式 pre-TTFT 沉默 → **不触发**(阶段 1 不开窗)
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {"s1": [("hang",), ("ok", "s1ok")], "s2": [("ok", "other")]}
+        )
+        mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S)
+        ctx = _ctx()
+        task = asyncio.ensure_future(mw(_req(ctx=ctx)))
+        await transport.entered["s1"].wait()
+        await asyncio.sleep(4 * _STALL_AFTER_S)  # 4× 余量确认 stall 窗管不到 pre-TTFT
+        assert transport.calls == ["s1"]
+        transport.release["s1"].set()
+        resp = await asyncio.wait_for(task, timeout=5)
+        assert resp.source_name == "s1"
+        stats = ctx.snapshot()
+        assert stats.hedges == 0 and stats.attempts == 1
+
+        # 格 3: 两者皆设 + 首 chunk 后沉默 → 阶段 2 stall 窗触发
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {"s1": [("chunk",), ("hang",), ("ok", "late")], "s2": [("ok", "fast")]}
+        )
+        mw, _, _ = _harness(
+            [s1, s2], transport, hedge_after_s=_STALL_AFTER_S, stall_after_s=_STALL_AFTER_S
+        )
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.source_name == "s2"
+        assert ctx.snapshot().hedges == 1
+
+        # 格 3b: 两者皆设 + pre-TTFT 挂起 → 阶段 1 AFTER_S 触发(两阶段各管一段)
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {"s1": [("hang",), ("ok", "late")], "s2": [("ok", "fast")]}
+        )
+        mw, _, _ = _harness(
+            [s1, s2], transport, hedge_after_s=_STALL_AFTER_S, stall_after_s=_STALL_AFTER_S
+        )
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.source_name == "s2"
+        assert ctx.snapshot().hedges == 1
+
+        # 格 4: 皆无 → 逐字单路(默认关闭回归门的单元格形态)
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport({"s1": [("hang",), ("ok", "solo")]})
+        mw, _, _ = _harness([s1, s2], transport)
+        ctx = _ctx()
+        task = asyncio.ensure_future(mw(_req(ctx=ctx)))
+        await transport.entered["s1"].wait()
+        transport.release["s1"].set()
+        resp = await asyncio.wait_for(task, timeout=5)
+        assert resp.content == "solo" and resp.source_name == "s1"
+        stats = ctx.snapshot()
+        assert stats.hedges == 0 and stats.attempts == 1
+        assert transport.ce_events == [None]  # 未配置 = 不观测增量 chunk
+
+    async def test_stall_only_non_stream_degrades_to_single_leg(self):
+        """仅 stall 已设 + `stream=False` → 逐字单路(中段沉默无非流式概念,入口门)。"""
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {"s1": [("hang",), ("ok", "x")], "s2": [("ok", "other")]}
+        )
+        mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S)
+        ctx = _ctx()
+        task = asyncio.ensure_future(mw(_req(stream=False, ctx=ctx)))
+        await transport.entered["s1"].wait()
+        await asyncio.sleep(4 * _STALL_AFTER_S)  # 阈值窗已过,仍不得有第二请求
+        assert transport.calls == ["s1"]
+        transport.release["s1"].set()
+        resp = await asyncio.wait_for(task, timeout=5)
+        assert resp.source_name == "s1"
+        stats = ctx.snapshot()
+        assert stats.hedges == 0 and stats.attempts == 1
+        assert transport.ce_events == [None]  # 非流式不建 content_event
+
+
+class TestStallHedgeWatchdogAndFailure:
+    """与看门狗的交互(设计 §5.2): 主路被掐死 ≠ 本轮失败,副路已在跑则等副路。"""
+
+    async def test_watchdog_kills_primary_hedge_leg_survives(self):
+        """主路首 chunk 后被看门狗掐死(Transient)、副路在途 → 副路结果裁定。
+
+        attempt 级记账照常(N9): 主路源熔断失败计数 +1、遥测主路 error 行
+        (TransientError 对象)+ 副路正常行;"先了结的是失败则等另一路"逐字复用。
+        """
+        emitter = RecordingEmitter()
+        gate = InMemoryGate(config=_BREAKER)
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {
+                "s1": [
+                    ("chunk",),
+                    (
+                        "fail_after",
+                        0.12,
+                        lambda: TransientError("inter_token timeout", source_name="s1"),
+                    ),
+                ],
+                "s2": [("ok_after", 0.2, "rescued")],
+            }
+        )
+        mw, _, _ = _harness(
+            [s1, s2], transport, stall_after_s=_STALL_AFTER_S, emitter=emitter, gate=gate
+        )
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.content == "rescued" and resp.source_name == "s2"
+        stats = ctx.snapshot()
+        assert stats.hedges == 1 and stats.hedge_won is True
+        # 掐死是 attempt 级失败: 熔断失败样本 +1(对冲输家取消不喂,此处是喂的——N9 不对称)
+        gate_s1 = gate._gates["s1"]
+        assert gate_s1.a0 + gate_s1.a1 == 1
+        # 遥测: 主路 error 行携 TransientError 对象(掐死),副路正常行
+        primary_row = next(r for r in emitter.rows if r["source"] == "s1")
+        hedge_row = next(r for r in emitter.rows if r["source"] == "s2")
+        assert isinstance(primary_row["error"], TransientError)
+        assert hedge_row["error"] is None
+
+    async def test_both_fail_counts_budget_once(self):
+        """主路看门狗掐死 + 副路 Transient: 两败只计一次预算(_combine_failures 复用)。
+
+        max_attempts=2 时恰进第二轮: 若两败计两次,第一轮即耗尽,第二轮的两路
+        调用根本不会发生。两败轮次无赢家 → generation 不记、stall 快照恒 None。
+        """
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {
+                "s1": [
+                    ("chunk",),
+                    ("fail_after", 0.1, lambda: TransientError("wd", source_name="s1")),
+                    ("chunk",),
+                    ("fail_after", 0.1, lambda: TransientError("wd", source_name="s1")),
+                ],
+                "s2": [
+                    ("fail_after", 0.12, lambda: TransientError("h", source_name="s2")),
+                    ("fail_after", 0.12, lambda: TransientError("h", source_name="s2")),
+                ],
+            }
+        )
+        mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S, max_attempts=2)
+        ctx = _ctx()
+        with pytest.raises(AllSourcesExhausted) as ei:
+            await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert ei.value.reason == "retry_exhausted"
+        assert transport.calls == ["s1", "s2", "s1", "s2"]  # 两败计一次才有第二轮
+        stats = ctx.snapshot()
+        assert stats.hedges == 2  # 两败轮次路数照登(实际并发发出即计)
+        assert stats.hedge_won is False
+        assert stats.generation_ms == 0  # 无赢家: 裸生成时间无归属不记
+        assert stats.stall_ms is None and stats.stall_count is None
+
+
+class TestStallHedgeCancellation:
+    """取消穿透(计划取消铁律): stall 等待中外部取消,waiter 收口无残留。"""
+
+    async def test_external_cancel_during_stall_wait_penetrates(self):
+        """阶段 2 循环内外部 `task.cancel()`: CancelledError 穿透、无残留任务。
+
+        阈值取 5.0s 宽窗: 取消确定落在 stall 等待内(阶段 2 循环),不依赖时序。
+        对冲尚未发出(宽窗未到期)→ 只有原路被取消,permit 释放。
+        """
+        before = asyncio.all_tasks()
+        emitter = RecordingEmitter()
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {"s1": [("chunk",), ("hang",), ("ok", "x")], "s2": [("ok", "y")]}
+        )
+        mw, limiter, _ = _harness([s1, s2], transport, stall_after_s=5.0, emitter=emitter)
+        task = asyncio.ensure_future(mw(_req(ctx=_ctx())))
+        await transport.entered["s1"].wait()  # 原路已交付首 chunk 并挂起: 编排处于阶段 2
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert transport.calls == ["s1"]  # 宽窗未到期,对冲从未发出
+        assert {r["source"]: r["error"] for r in emitter.rows} == {"s1": "cancelled"}
+        assert (await limiter.source_stats("s1")).inflight == 0
+        assert asyncio.all_tasks() == before  # waiter/原路全收口,无残留
+
+
+class TestStallHedgeEventDiscipline:
+    """Event 并发纪律(N3③): 对冲路 content_event 恒 None(v1 不梯次)。"""
+
+    async def test_hedge_leg_content_event_always_none(self):
+        """原路(流式 + stall 已设)收真事件,对冲路恒 None——逐次身份记录断言。"""
+        s1, s2 = _src("s1"), _src("s2")
+        transport = StallHedgeTransport(
+            {"s1": [("chunk",), ("hang",), ("ok", "late")], "s2": [("ok", "win")]}
+        )
+        mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S)
+        ctx = _ctx()
+        resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
+        assert resp.source_name == "s2"
+        assert transport.calls == ["s1", "s2"]
+        # 原路携带首 token + content 双观测;对冲路两事件恒 None(N3③,同 H5 首 token 约定)
+        assert [e is not None for e in transport.ft_events] == [True, False]
+        assert [e is not None for e in transport.ce_events] == [True, False]