feat: feed selector health and demote in-call failed sources

RetryMW keeps a per-call failure map (local, never instance state):
a source failing twice in one call yields to the next candidate.
Attempt outcomes feed OutcomeAwareSelector behind a swallow-and-warn
guard; ResultInvalid and provider-rejected paths record success with
count_attempt=False so the breaker window stays clean. Same accounting
applied in EmbeddingClient.

src/polygateway/embedding.py

@@ -303,7 +303,8 @@ class EmbeddingClient:
         """终态异常的门控写回(与 RetryMW 同口径): 坏结果/网关健康拒绝 ≠ 坏服务
         → 记成功;网关没响应的拒绝若持探针则归还。"""
         if isinstance(exc, ResultInvalidError) or exc.status_code is not None:
-            await self._record_quietly(self._breaker.record_success(entry))
+            # M2.5 §3.1: 记成功但不计失败率窗口样本(坏结果/坏请求 ≠ 坏服务)
+            await self._record_quietly(self._breaker.record_success(entry, count_attempt=False))
         elif entry.is_probe:
             await self._record_quietly(self._breaker.release_probe(entry))
 

src/polygateway/middleware/breaker.py

@@ -25,9 +25,11 @@ class BreakerGate:
         except Exception as exc:
             raise GovernanceBackendError(f"熔断后端故障(try_enter): {exc}") from exc
 
-    async def record_success(self, entry: GateDecision) -> GateUpdate:
+    async def record_success(
+        self, entry: GateDecision, *, count_attempt: bool = True
+    ) -> GateUpdate:
         try:
-            return await self._gate.record_success(entry)
+            return await self._gate.record_success(entry, count_attempt=count_attempt)
         except GovernanceBackendError:
             raise
         except Exception as exc:

src/polygateway/middleware/retry.py

@@ -32,6 +32,7 @@ from polygateway.errors import (
 )
 from polygateway.middleware.breaker import BreakerGate
 from polygateway.middleware.ratelimit import QuotaGate
+from polygateway.ports import OutcomeAwareSelector
 from polygateway.sources import SourceCooldownMemo
 from polygateway.streaming import StreamLivenessTimeout
 from polygateway.types import LLMResponse
@@ -72,6 +73,16 @@ def backoff_delay(
     return max(delay, retry_after)
 
 
+def _demote_call_failures(
+    ordered: list[SourceConfig], attempt_fails: dict[str, int]
+) -> list[SourceConfig]:
+    """调用内降权(设计 §3.3): 本调用失败 ≥2 次的源移尾;全失败则保持原序。"""
+    demoted = [s for s in ordered if attempt_fails.get(s.name, 0) >= 2]
+    if not demoted or len(demoted) == len(ordered):
+        return ordered
+    return [s for s in ordered if attempt_fails.get(s.name, 0) < 2] + demoted
+
+
 def _failure_reason(exc: PolyGatewayError) -> str:
     """失败原因归类(CHS governance.py:169 同款)。"""
     if isinstance(exc, SourceDeadError):
@@ -124,6 +135,8 @@ class RetryMW:
         self._bp = backpressure
         self._quota_full = quota_full
         self._memo = cooldown_memo or SourceCooldownMemo(now=now)
+        # M2.5: 选源器可选健康喂数端口,构造期 isinstance 判定一次(设计 §3.2)
+        self._outcome_sink = selector if isinstance(selector, OutcomeAwareSelector) else None
         self._emitter = emitter
         self._now = now
         self._sleep = sleep
@@ -135,13 +148,15 @@ class RetryMW:
             raise AllSourcesExhausted(scope=self._scope, reason="no_sources", retry_after_s=0.0)
         fails = 0
         reasons: dict[str, str] = {}
+        # 调用内失败计数(设计 §3.3): 局部状态,调用结束即弃;严禁实例属性(并发共享)
+        attempt_fails: dict[str, int] = {}
         entered_at = self._now()  # 调用级累计计时,循环内不重置(CHS governance.py:207)
         while True:
-            picked, gate_rejections = await self._pick_runnable(reasons)
+            picked, gate_rejections = await self._pick_runnable(reasons, attempt_fails)
             if picked is None:
                 await self._on_no_runnable(gate_rejections, reasons, entered_at)
                 continue
-            outcome = await self._attempt(request, *picked, reasons)
+            outcome = await self._attempt(request, *picked, reasons, attempt_fails)
             if isinstance(outcome, LLMResponse):
                 return outcome
             fails += 1
@@ -158,11 +173,12 @@ class RetryMW:
     # —— 选源与准入(CHS _pick_runnable 120-167)——
 
     async def _pick_runnable(
-        self, reasons: dict[str, str]
+        self, reasons: dict[str, str], attempt_fails: dict[str, int]
     ) -> tuple[tuple[SourceConfig, Permit, GateDecision] | None, int]:
         stats = {s.name: await self._quota.stats(s) for s in self._sources}
         gate_rejections = 0
-        for cand in self._selector.order(self._sources, stats):
+        ordered = _demote_call_failures(self._selector.order(self._sources, stats), attempt_fails)
+        for cand in ordered:
             if self._memo.active(cand.name):
                 # 冷却备忘跳过也计入拒绝数,保住 circuit_open 判据(CHS 同款)
                 gate_rejections += 1
@@ -228,6 +244,7 @@ class RetryMW:
         permit: Permit,
         entry: GateDecision,
         reasons: dict[str, str],
+        attempt_fails: dict[str, int],
     ) -> LLMResponse | _Failed:
         call_id = str(uuid.uuid4())
         started = self._now()
@@ -243,6 +260,7 @@ class RetryMW:
             actual = result.prompt_tokens + result.completion_tokens
             await self._record_quietly(self._breaker.record_success(entry))
             await self._record_quietly(self._quota.mark_progress())
+            self._feed_outcome(source.name, ok=True)
             response = self._build_response(source, result, call_id, started)
             await self._emit(request, source, call_id, started, response=response)
             return response
@@ -251,8 +269,8 @@ class RetryMW:
             await self._emit(request, source, call_id, started, error=exc)
             raise
         except ResultInvalidError as exc:
-            # 坏结果 ≠ 坏服务: 熔断记成功,异常上抛消耗业务失败预算(§6.3)
+            # 坏结果 ≠ 坏服务: 熔断记成功但不计窗口样本,亦不喂健康分(M2.5 §3.1)
-            await self._record_quietly(self._breaker.record_success(entry))
+            await self._record_quietly(self._breaker.record_success(entry, count_attempt=False))
             await self._emit(request, source, call_id, started, error=exc)
             raise
         except asyncio.CancelledError:
@@ -264,6 +282,8 @@ class RetryMW:
             dead = isinstance(exc, SourceDeadError)
             reason = _failure_reason(exc)
             reasons[source.name] = reason
+            attempt_fails[source.name] = attempt_fails.get(source.name, 0) + 1
+            self._feed_outcome(source.name, ok=False)
             await self._record_quietly(self._breaker.record_failure(entry, reason, dead))
             if not dead:
                 actual = source.est_tokens  # 保守: 失败请求可能已被网关计费(CHS 同款)
@@ -277,8 +297,8 @@ class RetryMW:
     ) -> None:
         provider_responded = exc.source_name == source.name and exc.status_code is not None
         if provider_responded:
-            # 网关健康地拒了坏请求
+            # 网关健康地拒了坏请求: 记成功但不计窗口样本(M2.5 §3.1)
-            await self._record_quietly(self._breaker.record_success(entry))
+            await self._record_quietly(self._breaker.record_success(entry, count_attempt=False))
         elif entry.is_probe:
             await self._record_quietly(self._breaker.release_probe(entry))
 
@@ -296,6 +316,15 @@ class RetryMW:
         except GovernanceBackendError as exc:
             logger.warning("治理记账写回降级(不冒泡): {}", exc)
 
+    def _feed_outcome(self, source_name: str, ok: bool) -> None:
+        """健康喂数降级执行: 选源器异常不得打断真实成功/失败的主路径(设计 §4)。"""
+        if self._outcome_sink is None:
+            return
+        try:
+            self._outcome_sink.record_outcome(source_name, ok)
+        except Exception as exc:
+            logger.warning("选源健康喂数失败(降级不冒泡): {}", exc)
+
     # —— 辅助 ——
 
     def _backoff_delay(self, fails: int, exc: PolyGatewayError) -> float:

src/polygateway/ports.py

@@ -152,7 +152,9 @@ class ProviderGate(Protocol):
 
     async def try_enter(self, source_name: str, owner: str) -> GateDecision: ...
 
-    async def record_success(self, entry: GateDecision) -> GateUpdate: ...
+    async def record_success(
+        self, entry: GateDecision, *, count_attempt: bool = True
+    ) -> GateUpdate: ...
 
     async def record_failure(
         self, entry: GateDecision, reason: str, force_open: bool

tests/unit/test_retry.py

@@ -97,6 +97,7 @@ def _harness(
     quota_full="wait",
     global_limits=_NO_GLOBAL,
     rng=lambda: 0.0,
+    selector=None,
 ):
     clock = clock or FakeClock()
     limiter = InMemoryLimiter(
@@ -112,7 +113,7 @@ def _harness(
     mw = RetryMW(
         scope="llm",
         sources=sources,
-        selector=RoundRobinSelector(),
+        selector=selector if selector is not None else RoundRobinSelector(),
         limiter=limiter,
         gate=gate,
         transport=transport,
@@ -351,3 +352,94 @@ class TestCancellation:
         # 探针租约已归还: 下一 caller 立即拿到探针而非等租约过期
         nxt = await gate.try_enter("a", "w2")
         assert nxt.allowed and nxt.is_probe
+
+
+class StaticSelector:
+    """固定配置序,隔离测试调用内降权(不带 record_outcome)。"""
+
+    def order(self, sources, stats):
+        return list(sources)
+
+
+class RecordingSelector(StaticSelector):
+    def __init__(self):
+        self.outcomes = []
+
+    def record_outcome(self, source_name, ok):
+        self.outcomes.append((source_name, ok))
+
+
+class ExplodingSelector(StaticSelector):
+    def record_outcome(self, source_name, ok):
+        raise RuntimeError("sink boom")
+
+
+class TestM25Orchestration:
+    """M2.5 设计 §3.3: 调用内失败降权 + 健康喂数(先红后绿)。"""
+
+    async def test_failed_source_demoted_after_two_strikes(self):
+        # 失败 1 次仍首选(原地退避重试);失败 2 次让位次优源
+        mw, _, _, transport, _, _ = _harness(
+            [_src("a"), _src("b")],
+            [TransientError("1"), TransientError("2"), _ok()],
+            selector=StaticSelector(),
+        )
+        resp = await mw(_REQ)
+        assert [n for n, _ in transport.calls] == ["a", "a", "b"]
+        assert resp.source_name == "b"
+
+    async def test_attempt_fails_reset_between_calls(self):
+        mw, _, _, transport, _, _ = _harness(
+            [_src("a"), _src("b")],
+            [TransientError("1"), TransientError("2"), _ok(), _ok()],
+            selector=StaticSelector(),
+        )
+        await mw(_REQ)
+        await mw(_REQ)  # 新调用状态清零: 回到首选 a
+        assert [n for n, _ in transport.calls] == ["a", "a", "b", "a"]
+
+    async def test_outcome_feeding_success_and_transient(self):
+        sel = RecordingSelector()
+        mw, _, _, _, _, _ = _harness(
+            [_src("a"), _src("b")], [TransientError("1"), _ok()], selector=sel
+        )
+        await mw(_REQ)
+        assert sel.outcomes == [("a", False), ("a", True)]
+
+    async def test_outcome_feeding_source_dead(self):
+        sel = RecordingSelector()
+        mw, _, _, _, _, _ = _harness(
+            [_src("a"), _src("b")], [SourceDeadError("401"), _ok()], selector=sel
+        )
+        await mw(_REQ)
+        assert sel.outcomes == [("a", False), ("b", True)]
+
+    async def test_result_invalid_and_rejected_not_fed(self):
+        sel = RecordingSelector()
+        mw, _, _, _, _, _ = _harness(
+            [_src("a")], [ResultInvalidError("bad", raw_text="x")], selector=sel
+        )
+        with pytest.raises(ResultInvalidError):
+            await mw(_REQ)
+        sel2 = RecordingSelector()
+        mw2, _, _, _, _, _ = _harness(
+            [_src("a")],
+            [RequestRejectedError("400", source_name="a", status_code=400)],
+            selector=sel2,
+        )
+        with pytest.raises(RequestRejectedError):
+            await mw2(_REQ)
+        assert sel.outcomes == [] and sel2.outcomes == []
+
+    async def test_outcome_sink_exception_swallowed(self):
+        mw, _, _, _, _, _ = _harness([_src("a")], [_ok()], selector=ExplodingSelector())
+        resp = await mw(_REQ)
+        assert resp.content == "ok"  # 喂数异常不得打断真实成功返回
+
+    async def test_result_invalid_gate_window_untouched(self):
+        # 坏结果 ≠ 坏服务: count_attempt=False,失败率窗口 attempts 不得增长
+        mw, _, gate, _, _, _ = _harness([_src("a")], [ResultInvalidError("bad", raw_text="x")])
+        with pytest.raises(ResultInvalidError):
+            await mw(_REQ)
+        g = gate._gates["a"]
+        assert g.a0 + g.a1 == 0