fix(gateway): 429 同 key 只再试一次，换 key 前拉长退避

backend/packages/framework/src/windup_framework/gateway/chat.py

@@ -17,7 +17,7 @@
 from windup_framework.gateway.budget import AttemptBudget
 from windup_framework.gateway.circuit import CircuitBreaker
 from windup_framework.gateway.context import current_call_context
-from windup_framework.gateway.policy import decide
+from windup_framework.gateway.policy import decide, rate_limit_wait_s
 from windup_framework.gateway.routes import (
     GatewayRoute,
     config_for_route,
@@ -30,8 +30,6 @@
 from windup_framework.gateway.types import Family, NextStep, Scene
 
 _CIRCUIT = CircuitBreaker()
-_DEFAULT_RETRY_AFTER_S = 2.0
-_SLEEP_CAP_S = 30.0
 
 
 _ERROR_MESSAGE_LIMIT = 2_000
@@ -455,19 +453,31 @@ def fail(http_status: int | None) -> None:
                         break
                     if step is NextStep.FALLBACK_KEY:
                         if has_next_route:
+                            nxt = self._routes[route_index + 1]
+                            time.sleep(
+                                rate_limit_wait_s(
+                                    retry_count=retry_count,
+                                    retry_after_s=result.retry_after_s,
+                                )
+                            )
                             fallback_used = True
-                            route_reason_override = "key_rate_limit"
+                            if nxt.base_url_id != route.base_url_id:
+                                self._circuit.open("base_url:" + route.base_url_id)
+                                route_reason_override = "base_url_unreached"
+                            else:
+                                route_reason_override = "key_rate_limit"
                             switch_to_next_route = True
                             break
+                        self._circuit.open("aggregator")
                         fail(last_http_status)
                     if step is NextStep.RETRY_SAME:
                         if error_type is ModelErrorType.RATE_LIMIT:
-                            wait = (
-                                result.retry_after_s
-                                if result.retry_after_s is not None
-                                else _DEFAULT_RETRY_AFTER_S
+                            time.sleep(
+                                rate_limit_wait_s(
+                                    retry_count=retry_count,
+                                    retry_after_s=result.retry_after_s,
+                                )
                             )
-                            time.sleep(min(wait, _SLEEP_CAP_S))
                         retry_count += 1
                         continue
                     if step is NextStep.FALLBACK:

backend/packages/framework/src/windup_framework/gateway/image.py

@@ -10,7 +10,7 @@
 from windup_framework.gateway.budget import AttemptBudget
 from windup_framework.gateway.circuit import CircuitBreaker
 from windup_framework.gateway.context import current_call_context
-from windup_framework.gateway.policy import decide
+from windup_framework.gateway.policy import decide, rate_limit_wait_s
 from windup_framework.gateway.registry import ModelRegistry, RegistryError
 from windup_framework.gateway.routes import (
     GatewayRoute,
@@ -31,8 +31,6 @@
 from windup_framework.gateway.types import NextStep, Scene
 
 _CIRCUIT = CircuitBreaker()
-_DEFAULT_RETRY_AFTER_S = 2.0
-_SLEEP_CAP_S = 30.0
 
 
 def _utc_now() -> str:
@@ -347,19 +345,31 @@ def fail(http_status: int | None) -> None:
                         break
                     if step is NextStep.FALLBACK_KEY:
                         if has_next_route:
+                            nxt = routes[route_index + 1]
+                            time.sleep(
+                                rate_limit_wait_s(
+                                    retry_count=retry_count,
+                                    retry_after_s=result.retry_after_s,
+                                )
+                            )
                             fallback_used = True
-                            route_reason_override = "key_rate_limit"
+                            if nxt.base_url_id != route.base_url_id:
+                                self._circuit.open("base_url:" + route.base_url_id)
+                                route_reason_override = "base_url_unreached"
+                            else:
+                                route_reason_override = "key_rate_limit"
                             switch_to_next_route = True
                             break
+                        self._circuit.open("aggregator")
                         fail(last_http_status)
                     if step is NextStep.RETRY_SAME:
                         if error_type is ModelErrorType.RATE_LIMIT:
-                            wait = (
-                                result.retry_after_s
-                                if result.retry_after_s is not None
-                                else _DEFAULT_RETRY_AFTER_S
+                            time.sleep(
+                                rate_limit_wait_s(
+                                    retry_count=retry_count,
+                                    retry_after_s=result.retry_after_s,
+                                )
                             )
-                            time.sleep(min(wait, _SLEEP_CAP_S))
                         retry_count += 1
                         if error_type is ModelErrorType.UNREACHED:
                             resend_spent = 1

backend/packages/framework/src/windup_framework/gateway/policy.py

@@ -3,6 +3,17 @@
 from windup_common.enums.model import ModelErrorType
 from windup_framework.gateway.types import NextStep
 
+RATE_LIMIT_BACKOFF_BASE_S = 8.0
+RATE_LIMIT_SLEEP_CAP_S = 60.0
+
+
+def rate_limit_wait_s(*, retry_count: int, retry_after_s: float | None) -> float:
+    """429 等待：同 key 再试 8s，换 key 16s。Retry-After 作下限，封顶 60s。"""
+    wait = RATE_LIMIT_BACKOFF_BASE_S * (2 ** retry_count)
+    if retry_after_s is not None:
+        wait = max(wait, retry_after_s)
+    return min(wait, RATE_LIMIT_SLEEP_CAP_S)
+
 
 def decide(
     *,
@@ -28,7 +39,7 @@ def decide(
         return NextStep.FAIL
     if error_type is ModelErrorType.UNREACHED:
         return NextStep.OPEN_AGGREGATOR
-    if error_type is ModelErrorType.RATE_LIMIT and retry_count < 2:
+    if error_type is ModelErrorType.RATE_LIMIT and retry_count == 0:
         return NextStep.RETRY_SAME
     if error_type is ModelErrorType.RATE_LIMIT:
         return NextStep.FALLBACK_KEY

backend/packages/framework/src/windup_framework/gateway/video.py

@@ -11,7 +11,7 @@
 from windup_framework.gateway.budget import AttemptBudget
 from windup_framework.gateway.context import current_call_context
 from windup_framework.gateway.image import _CIRCUIT
-from windup_framework.gateway.policy import decide
+from windup_framework.gateway.policy import decide, rate_limit_wait_s
 from windup_framework.gateway.registry import ModelRegistry, RegistryError
 from windup_framework.gateway.routes import (
     GatewayRoute,
@@ -31,9 +31,6 @@
 )
 from windup_framework.gateway.types import AdapterResult, NextStep, Scene
 
-_DEFAULT_RETRY_AFTER_S = 2.0
-_SLEEP_CAP_S = 30.0
-
 
 @dataclass(frozen=True)
 class SubmittedVideoJob:
@@ -389,19 +386,31 @@ def fail(http_status: int | None) -> None:
                         if bound_job_id is not None:
                             fail(last_http_status)
                         if has_next_route:
+                            nxt = routes[route_index + 1]
+                            time.sleep(
+                                rate_limit_wait_s(
+                                    retry_count=retry_count,
+                                    retry_after_s=result.retry_after_s,
+                                )
+                            )
                             fallback_used = True
-                            route_reason_override = "key_rate_limit"
+                            if nxt.base_url_id != route.base_url_id:
+                                self._circuit.open("base_url:" + route.base_url_id)
+                                route_reason_override = "base_url_unreached"
+                            else:
+                                route_reason_override = "key_rate_limit"
                             switch_to_next_route = True
                             break
+                        self._circuit.open("aggregator")
                         fail(last_http_status)
                     if step is NextStep.RETRY_SAME:
                         if error_type is ModelErrorType.RATE_LIMIT:
-                            wait = (
-                                result.retry_after_s
-                                if result.retry_after_s is not None
-                                else _DEFAULT_RETRY_AFTER_S
+                            time.sleep(
+                                rate_limit_wait_s(
+                                    retry_count=retry_count,
+                                    retry_after_s=result.retry_after_s,
+                                )
                             )
-                            time.sleep(min(wait, _SLEEP_CAP_S))
                         retry_count += 1
                         if error_type is ModelErrorType.UNREACHED:
                             resend_spent = 1

backend/tests/test_gateway_chat.py

@@ -73,7 +73,7 @@ def test_chat_gateway_switches_key_after_429(monkeypatch, caplog):
     monkeypatch.setattr("windup_framework.gateway.chat.time.sleep", lambda _: None)
     caplog.set_level(logging.INFO, logger="windup.gateway")
     rate = ChatAdapterResult(ok=False, error_type=ModelErrorType.RATE_LIMIT, http_status=429)
-    key_a = FakeChatAdapter({"gpt-4o-mini": [rate, rate, rate]})
+    key_a = FakeChatAdapter({"gpt-4o-mini": [rate, rate]})
     key_b = FakeChatAdapter({"gpt-4o-mini": [OK]})
     cfg = AIProviderSettings(
         model="gpt-4o-mini",
@@ -90,7 +90,7 @@ def test_chat_gateway_switches_key_after_429(monkeypatch, caplog):
     )
 
     assert gw.invoke([{"role": "user", "content": "ping"}]) == "pong"
-    assert key_a.calls == ["gpt-4o-mini"] * 3
+    assert key_a.calls == ["gpt-4o-mini"] * 2
     assert key_b.calls == ["gpt-4o-mini"]
     records = [json.loads(r.message) for r in caplog.records if r.name == "windup.gateway"]
     success = [r for r in records if r.get("outcome") in ("success", "fallback_success")]

backend/tests/test_gateway_image.py

@@ -103,22 +103,53 @@ def test_429_does_not_switch_model_when_only_one_key(monkeypatch):
     monkeypatch.setattr("windup_framework.gateway.image.time.sleep", lambda _: None)
     rate = AdapterResult(ok=False, error_type=ModelErrorType.RATE_LIMIT, http_status=429)
     ad = FakeImageAdapter({
-        "gemini-2.5-flash-image": [rate, rate, rate],
+        "gemini-2.5-flash-image": [rate, rate],
         "gemini-2.5-flash-image-alt": [PNG],
     })
     gw = _make_gw(ad, image_fallbacks="gemini-2.5-flash-image-alt")
     with pytest.raises(RuntimeError, match="429"):
         gw.gen_image("p", [])
-    assert ad.calls == ["gemini-2.5-flash-image"] * 3
+    assert ad.calls == ["gemini-2.5-flash-image"] * 2
     assert "gemini-2.5-flash-image-alt" not in ad.calls
 
 
+def test_429_same_key_retries_once_then_opens_aggregator(monkeypatch):
+    slept: list[float] = []
+    monkeypatch.setattr("windup_framework.gateway.image.time.sleep", slept.append)
+    rate = AdapterResult(ok=False, error_type=ModelErrorType.RATE_LIMIT, http_status=429)
+    ad = FakeImageAdapter({"gemini-2.5-flash-image": [rate, rate]})
+    br = CircuitBreaker()
+    gw = _make_gw(ad, circuit=br)
+    with pytest.raises(RuntimeError, match="429"):
+        gw.gen_image("p", [])
+    assert ad.calls == ["gemini-2.5-flash-image"] * 2
+    assert slept == [8.0]
+    assert br.is_open("aggregator")
+
+
+def test_429_backoff_waits_at_least_retry_after(monkeypatch):
+    slept: list[float] = []
+    monkeypatch.setattr("windup_framework.gateway.image.time.sleep", slept.append)
+    rate = AdapterResult(
+        ok=False,
+        error_type=ModelErrorType.RATE_LIMIT,
+        http_status=429,
+        retry_after_s=10.0,
+    )
+    ad = FakeImageAdapter({"gemini-2.5-flash-image": [rate, rate]})
+    gw = _make_gw(ad)
+    with pytest.raises(RuntimeError, match="429"):
+        gw.gen_image("p", [])
+    assert slept == [10.0]
+
+
 def test_429_switches_key_on_same_base_url_before_model(monkeypatch, caplog):
-    monkeypatch.setattr("windup_framework.gateway.image.time.sleep", lambda _: None)
+    slept: list[float] = []
+    monkeypatch.setattr("windup_framework.gateway.image.time.sleep", slept.append)
     caplog.set_level(logging.INFO, logger="windup.gateway")
     rate = AdapterResult(ok=False, error_type=ModelErrorType.RATE_LIMIT, http_status=429)
     key_a = FakeImageAdapter({
-        "gemini-2.5-flash-image": [rate, rate, rate],
+        "gemini-2.5-flash-image": [rate, rate],
         "gemini-2.5-flash-image-alt": [PNG],
     })
     key_b = FakeImageAdapter({"gemini-2.5-flash-image": [PNG]})
@@ -139,9 +170,10 @@ def test_429_switches_key_on_same_base_url_before_model(monkeypatch, caplog):
     )
 
     assert gw.gen_image("p", []).startswith(b"\x89PNG")
-    assert key_a.calls == ["gemini-2.5-flash-image"] * 3
+    assert key_a.calls == ["gemini-2.5-flash-image"] * 2
     assert key_b.calls == ["gemini-2.5-flash-image"]
     assert "gemini-2.5-flash-image-alt" not in key_a.calls
+    assert slept == [8.0, 16.0]
 
     records = [json.loads(r.message) for r in caplog.records if r.name == "windup.gateway"]
     success = [r for r in records if r.get("outcome") in ("success", "fallback_success")]
@@ -153,6 +185,41 @@ def test_429_switches_key_on_same_base_url_before_model(monkeypatch, caplog):
     assert line["api_key_id"].endswith("key1")
 
 
+def test_429_exhausted_keys_switches_backup_entry(monkeypatch, caplog):
+    monkeypatch.setattr("windup_framework.gateway.image.time.sleep", lambda _: None)
+    caplog.set_level(logging.INFO, logger="windup.gateway")
+    rate = AdapterResult(ok=False, error_type=ModelErrorType.RATE_LIMIT, http_status=429)
+    key_a = FakeImageAdapter({"gemini-2.5-flash-image": [rate, rate]})
+    backup = FakeImageAdapter({"gemini-2.5-flash-image": [PNG]})
+    cfg = AIProviderSettings(
+        image_model="gemini-2.5-flash-image",
+        route_primary_name="primary",
+        route_primary_base_url="https://api.qnaigc.com/v1",
+        route_primary_api_key="key-a",
+        route_fallback_name="backup",
+        route_fallback_base_url="https://backup.example.com/v1",
+        route_fallback_api_key="key-c",
+    )
+    br = CircuitBreaker()
+    gw = ImageGateway(
+        ModelRegistry.from_settings(cfg),
+        key_a,
+        br,
+        cfg,
+        route_adapters={"primary.key0": key_a, "backup.key0": backup},
+    )
+
+    assert gw.gen_image("p", []).startswith(b"\x89PNG")
+    assert key_a.calls == ["gemini-2.5-flash-image"] * 2
+    assert backup.calls == ["gemini-2.5-flash-image"]
+    assert br.is_open("base_url:primary")
+    records = [json.loads(r.message) for r in caplog.records if r.name == "windup.gateway"]
+    success = [r for r in records if r.get("outcome") in ("success", "fallback_success")]
+    line = success[-1]
+    assert line["route_reason"] == "base_url_unreached"
+    assert line["base_url_id"] == "backup"
+
+
 def test_522_skips_remaining_keys_on_same_url(caplog):
     caplog.set_level(logging.INFO, logger="windup.gateway")
     key_a = FakeImageAdapter({

backend/tests/test_gateway_policy.py

@@ -3,7 +3,7 @@
 
 from windup_common.enums.model import ModelErrorType
 from windup_framework.gateway.circuit import CircuitBreaker
-from windup_framework.gateway.policy import decide
+from windup_framework.gateway.policy import decide, rate_limit_wait_s
 from windup_framework.gateway.types import NextStep
 
 
@@ -12,10 +12,23 @@ def test_522_retries_once_then_opens_aggregator():
     assert decide(error_type=ModelErrorType.UNREACHED, retry_count=1, has_job_id=False) is NextStep.OPEN_AGGREGATOR
 
 
-def test_429_retries_twice_then_fallback_key():
+def test_429_retries_once_then_fallback_key():
     assert decide(error_type=ModelErrorType.RATE_LIMIT, retry_count=0, has_job_id=False) is NextStep.RETRY_SAME
-    assert decide(error_type=ModelErrorType.RATE_LIMIT, retry_count=1, has_job_id=False) is NextStep.RETRY_SAME
-    assert decide(error_type=ModelErrorType.RATE_LIMIT, retry_count=2, has_job_id=False) is NextStep.FALLBACK_KEY
+    assert decide(error_type=ModelErrorType.RATE_LIMIT, retry_count=1, has_job_id=False) is NextStep.FALLBACK_KEY
+
+
+def test_429_backoff_is_exponential():
+    assert rate_limit_wait_s(retry_count=0, retry_after_s=None) == 8.0
+    assert rate_limit_wait_s(retry_count=1, retry_after_s=None) == 16.0
+
+
+def test_429_backoff_uses_retry_after_as_floor():
+    assert rate_limit_wait_s(retry_count=0, retry_after_s=10.0) == 10.0
+    assert rate_limit_wait_s(retry_count=1, retry_after_s=3.0) == 16.0
+
+
+def test_429_backoff_caps_at_60s():
+    assert rate_limit_wait_s(retry_count=0, retry_after_s=100.0) == 60.0
 
 
 def test_520_never_retries():

backend/tests/test_gateway_video.py

@@ -107,7 +107,7 @@ def test_submit_429_switches_key_on_same_base_url(monkeypatch):
     rate = AdapterResult(ok=False, error_type=ModelErrorType.RATE_LIMIT, http_status=429)
     key_a = FakeVideoAdapter(
         submits={
-            "kling-v2-5-turbo": [rate, rate, rate],
+            "kling-v2-5-turbo": [rate, rate],
             "kling-v2-6": [AdapterResult(ok=True, job_id="wrong", maybe_billed=True)],
         },
         follows={},
@@ -136,7 +136,7 @@ def test_submit_429_switches_key_on_same_base_url(monkeypatch):
     )
 
     assert gw.i2v(b"frame", "walk").startswith(b"\x00\x00\x00\x18ftyp")
-    assert key_a.submit_models == ["kling-v2-5-turbo"] * 3
+    assert key_a.submit_models == ["kling-v2-5-turbo"] * 2
     assert key_b.submit_models == ["kling-v2-5-turbo"]
     assert "kling-v2-6" not in key_a.submit_models
 
@@ -172,7 +172,8 @@ def test_timeout_does_not_submit_fallback():
     "error_type",
     [ModelErrorType.RATE_LIMIT, ModelErrorType.INVALID_RESPONSE],
 )
-def test_follow_fallback_without_upstream_fail_does_not_open_second_job(error_type):
+def test_follow_fallback_without_upstream_fail_does_not_open_second_job(error_type, monkeypatch):
+    monkeypatch.setattr("windup_framework.gateway.video.time.sleep", lambda _: None)
     follow_result = AdapterResult(
         ok=False,
         error_type=error_type,
@@ -190,7 +191,8 @@ def test_follow_fallback_without_upstream_fail_does_not_open_second_job(error_ty
     with pytest.raises(RuntimeError, match=error_type.value):
         _video_gw(ad).i2v(b"frame", "walk")
     assert ad.submit_models == ["kling-v2-5-turbo"]
-    assert ad.followed == ["j1", "j1", "j1"]
+    follows = 2 if error_type is ModelErrorType.RATE_LIMIT else 3
+    assert ad.followed == ["j1"] * follows
 
 
 def test_success_trace_has_phase_timings(caplog):