fix(throttle): reset consecutive_429s on non-rate-limit failure

In `release_failure`, the cascade counter wasn't reset, so a sequence
like 429 → 500 → 429 was treated as 2 consecutive 429s. The cascade
counter feeds AIMD's reduce-once-per-cascade logic; the second 429
should start a fresh cascade and trigger another concurrency reduction,
but currently doesn't.

Standalone bug surfaced during #575 investigation; not on the failure
path that drives the gate-trip outcome but worth fixing while we're
in this code.

Author: andreatgretel

packages/data-designer-engine/src/data_designer/engine/models/clients/throttle_manager.py

@@ -310,6 +310,10 @@ def release_failure(
         with self._lock:
             state = self._get_or_create_domain(provider_name, model_id, domain)
             state.in_flight = max(0, state.in_flight - 1)
+            # Non-rate-limit failure breaks the 429 cascade: a sequence like
+            # 429 → 500 → 429 should treat the second 429 as the start of a
+            # new cascade, not the third in a row.
+            state.consecutive_429s = 0
 
     # -------------------------------------------------------------------
     # Sync / async wrappers

packages/data-designer-engine/tests/engine/models/clients/test_throttle_manager.py

@@ -183,6 +183,24 @@ def test_failure_releases_slot_without_limit_change(manager: ThrottleManager) ->
     assert state.in_flight == 0
 
 
+def test_failure_resets_consecutive_429s_cascade(manager: ThrottleManager) -> None:
+    """Non-rate-limit failure breaks the 429 cascade so 429→500→429 isn't treated as 2-in-a-row.
+
+    The cascade counter feeds the AIMD reduce-once-per-cascade logic; if a
+    non-RL failure doesn't reset it, the subsequent 429 is treated as part of
+    the previous cascade and the limit isn't reduced when it should be.
+    """
+    manager.try_acquire(provider_name=PROVIDER, model_id=MODEL, domain=DOMAIN, now=0.0)
+    manager.release_rate_limited(provider_name=PROVIDER, model_id=MODEL, domain=DOMAIN, now=0.0)
+    state = manager.get_domain_state(PROVIDER, MODEL, DOMAIN)
+    assert state is not None
+    assert state.consecutive_429s == 1
+
+    manager.try_acquire(provider_name=PROVIDER, model_id=MODEL, domain=DOMAIN, now=0.0)
+    manager.release_failure(provider_name=PROVIDER, model_id=MODEL, domain=DOMAIN, now=0.0)
+    assert state.consecutive_429s == 0
+
+
 # --- Global cap ---