feat: exempt 429 pushback from retry budget with stall ceiling

Round 6 hit account-level rate throttling the concurrency AIMD cannot
absorb: at 26 req/min the gateway still returned 16% 429s and each one
burned a third of the retry budget. Retry-After-guided 429s now back
off without consuming attempts (gRPC pushback semantics); the retry
loop gains a per-call ceiling using the same dual-condition stall
verdict as quota-wait (local window exceeded AND no global progress).

tests/unit/test_retry.py

@@ -557,3 +557,47 @@ class TestDemotionInsertPosition:
         srcs = [_src("a"), _src("b"), _src("c")]
         out = _demote_call_failures(srcs, {"a": 2}, None)
         assert [s.name for s in out] == ["b", "c", "a"]
+
+
+class TestRateLimitPushback:
+    """迭代 5(设计 §3.38): 429 是服务端调度指令,不耗重试预算;时间上限兜底。"""
+
+    async def test_429_does_not_consume_retry_budget(self):
+        # 3 连 429 后成功——若 429 计预算,max_attempts=3 时第 4 次不会发生
+        mw, _, _, transport, sleep, _ = _harness(
+            [_src("a")],
+            [
+                TransientError("t1", status_code=429, retry_after_s=1.0),
+                TransientError("t2", status_code=429, retry_after_s=1.0),
+                TransientError("t3", status_code=429, retry_after_s=1.0),
+                _ok(),
+            ],
+        )
+        resp = await mw(_REQ)
+        assert resp.content == "ok"
+        assert len(transport.calls) == 4
+        assert len(sleep.delays) == 3  # 每次 429 仍按 Retry-After 退避
+
+    async def test_429_storm_bounded_by_stall_window(self):
+        # 持续 429 且时钟推进超 stall_window → stalled 兜底,不无限循环
+        clock = FakeClock()
+        script = [TransientError(str(i), status_code=429, retry_after_s=30.0) for i in range(99)]
+        mw, _, _, _, _, _ = _harness([_src("a")], script, clock=clock)
+
+        async def advancing_sleep(seconds):
+            clock.advance(seconds)
+
+        mw._sleep = advancing_sleep
+        with pytest.raises(AllSourcesExhausted) as ei:
+            await mw(_REQ)
+        assert ei.value.reason == "stalled"
+
+    async def test_non_429_transient_still_consumes_budget(self):
+        mw, _, _, transport, _, _ = _harness(
+            [_src("a")],
+            [TransientError("1"), TransientError("2"), TransientError("3")],
+        )
+        with pytest.raises(AllSourcesExhausted) as ei:
+            await mw(_REQ)
+        assert ei.value.reason == "retry_exhausted"
+        assert len(transport.calls) == 3