fix(agent): honor structured quota reset signals

agent/error_classifier.py

@@ -803,6 +803,7 @@ def classify_api_error(
         status_code = 429
     body = _extract_error_body(error)
     error_code = _extract_error_code(body)
+    response_headers = _extract_response_headers(error)
 
     # Build a comprehensive error message string for pattern matching.
     # str(error) alone may not include the body message (e.g. OpenAI SDK's
@@ -1047,6 +1048,7 @@ def classify_api_error(
             provider=provider_lower, model=model_lower,
             approx_tokens=approx_tokens, context_length=context_length,
             num_messages=num_messages,
+            response_headers=response_headers,
             result_fn=_result,
         )
         if classified is not None:
@@ -1204,6 +1206,7 @@ def _classify_by_status(
     approx_tokens: int,
     context_length: int,
     num_messages: int = 0,
+    response_headers=None,
     result_fn,
 ) -> Optional[ClassifiedError]:
     """Classify based on HTTP status code with message-aware refinement."""
@@ -1348,8 +1351,10 @@ def _classify_by_status(
             or "usage limit" in error_msg
             or "usage_limit_reached" in error_msg
         )
-        has_transient_signal = any(
-            p in error_msg for p in _USAGE_LIMIT_TRANSIENT_SIGNALS
+        has_transient_signal = _has_usage_limit_transient_signal(
+            error_msg,
+            body,
+            response_headers,
         )
         if has_usage_limit and not has_transient_signal:
             return result_fn(
@@ -1465,6 +1470,41 @@ def _classify_by_status(
     return None
 
 
+def _has_usage_limit_transient_signal(
+    error_msg: str,
+    body: dict,
+    response_headers,
+) -> bool:
+    """Return whether a usage-limit response identifies a reset window."""
+    if any(pattern in error_msg for pattern in _USAGE_LIMIT_TRANSIENT_SIGNALS):
+        return True
+
+    payloads = [body]
+    if isinstance(body, dict) and isinstance(body.get("error"), dict):
+        payloads.append(body["error"])
+    reset_fields = ("resets_in_seconds", "resets_at", "reset_at", "retry_after")
+    for payload in payloads:
+        if not isinstance(payload, dict):
+            continue
+        if any(
+            payload.get(field) is not None and payload.get(field) != ""
+            for field in reset_fields
+        ):
+            return True
+
+    if response_headers and hasattr(response_headers, "get"):
+        for header in (
+            "retry-after",
+            "Retry-After",
+            "x-ratelimit-reset",
+            "X-RateLimit-Reset",
+        ):
+            value = response_headers.get(header)
+            if value is not None and value != "":
+                return True
+    return False
+
+
 def _classify_402(error_msg: str, result_fn) -> ClassifiedError:
     """Disambiguate 402: billing exhaustion vs transient usage limit.
 
@@ -1985,6 +2025,21 @@ def _extract_error_body(error: Exception) -> dict:
     return {}
 
 
+def _extract_response_headers(error: Exception):
+    """Walk the error and its cause chain to find response headers."""
+    current = error
+    for _ in range(5):
+        response = getattr(current, "response", None)
+        headers = getattr(response, "headers", None)
+        if headers and hasattr(headers, "get"):
+            return headers
+        cause = getattr(current, "__cause__", None) or getattr(current, "__context__", None)
+        if cause is None or cause is current:
+            break
+        current = cause
+    return {}
+
+
 def _extract_error_code(body: dict) -> str:
     """Extract an error code string from the response body."""
     if not body:

tests/agent/test_error_classifier.py

@@ -1,5 +1,7 @@
 """Tests for agent.error_classifier — structured API error classification."""
 
+from types import SimpleNamespace
+
 import pytest
 from agent.error_classifier import (
     ClassifiedError,
@@ -17,10 +19,11 @@ from agent.error_classifier import (
 
 class MockAPIError(Exception):
     """Simulates an OpenAI SDK APIStatusError."""
-    def __init__(self, message, status_code=None, body=None):
+    def __init__(self, message, status_code=None, body=None, headers=None):
         super().__init__(message)
         self.status_code = status_code
         self.body = body or {}
+        self.response = SimpleNamespace(headers=headers or {})
 
 
 class MockTransportError(Exception):
@@ -302,6 +305,56 @@ class TestClassifyApiError:
         assert result.reason == FailoverReason.rate_limit
         assert result.retryable is True
 
+    @pytest.mark.parametrize(
+        ("reset_field", "reset_value"),
+        [
+            ("resets_in_seconds", 3600),
+            ("resets_at", "2026-08-24T10:00:00Z"),
+            ("reset_at", "2026-08-24T10:00:00Z"),
+            ("retry_after", 3600),
+        ],
+    )
+    def test_anthropic_429_usage_limit_with_structured_reset_stays_rate_limit(
+        self,
+        reset_field,
+        reset_value,
+    ):
+        e = MockAPIError(
+            "usage limit reached",
+            status_code=429,
+            body={
+                "error": {
+                    "type": "usage_limit_reached",
+                    "message": "Your account has reached its usage limit.",
+                    reset_field: reset_value,
+                }
+            },
+        )
+
+        result = classify_api_error(e, provider="anthropic", model="claude-opus-5")
+
+        assert result.reason == FailoverReason.rate_limit
+        assert result.retryable is True
+
+    @pytest.mark.parametrize("header", ["Retry-After", "x-ratelimit-reset"])
+    def test_anthropic_429_usage_limit_with_reset_header_stays_rate_limit(self, header):
+        e = MockAPIError(
+            "usage limit reached",
+            status_code=429,
+            body={
+                "error": {
+                    "type": "usage_limit_reached",
+                    "message": "Your account has reached its usage limit.",
+                }
+            },
+            headers={header: "3600"},
+        )
+
+        result = classify_api_error(e, provider="anthropic", model="claude-opus-5")
+
+        assert result.reason == FailoverReason.rate_limit
+        assert result.retryable is True
+
     def test_alibaba_rate_increased_too_quickly(self):
         """Alibaba/DashScope returns a unique throttling message.