feat: classify HTTP 429 as RateLimitExceededError with retry-after metadata and improve weighted routing resilience

- Anthropic connector: extract Retry-After headers, raise RateLimitExceededError for 429 responses
- Streaming error mapper: promote bare BackendError(429) to RateLimitExceededError
- Composite failure recovery: recycle candidates on transient failures instead of exhausting
- Add tests for rate limiting, streaming error propagation, and CBOR timestamp stability
- Remove obsolete test files for deleted domain modules

Author: Mateusz

src/connectors/anthropic.py

@@ -25,8 +25,10 @@
 )
 from src.core.common.exceptions import (
     AuthenticationError,
+    BackendError,
     ConfigurationError,
     InvalidRequestError,
+    RateLimitExceededError,
     ServiceUnavailableError,
 )
 from src.core.config.app_config import AppConfig
@@ -72,6 +74,37 @@
 _LLM_PROXY_CLIENT_HOST_KEY = "_llm_proxy_client_host"
 
 
+def _retry_after_metadata_from_httpx_headers(
+    headers: Any,
+) -> tuple[dict[str, Any], int | None]:
+    """Extract Retry-After for resilience (same ``details['headers']`` shape as OpenAI).
+
+    ``RateLimitErrorHandler`` reads ``details['headers']['retry-after']`` when
+    ``reset_at`` is not a usable wall-clock hint, so we populate that structure here.
+    """
+
+    details: dict[str, Any] = {}
+    reset_hint: int | None = None
+    retry_after: str | None = None
+    try:
+        if hasattr(headers, "get"):
+            got = headers.get("retry-after")
+            if got is not None:
+                retry_after = str(got).strip()
+        if not retry_after:
+            for key, value in headers.items():
+                if str(key).lower() == "retry-after":
+                    retry_after = str(value).strip()
+                    break
+        if retry_after:
+            details["headers"] = {"retry-after": retry_after}
+            with contextlib.suppress(ValueError, TypeError):
+                reset_hint = int(retry_after.split(",")[0].strip())
+    except Exception:
+        return {}, None
+    return details, reset_hint
+
+
 def _message_tool_calls(msg: Any) -> list[Any] | None:
     raw = (
         msg.get("tool_calls")
@@ -552,7 +585,7 @@ async def chat_completions(  # type: ignore[override]
                 ),
                 details={"connector": "anthropic"},
             )
-        return await self._chat_completions_canonical(request)
+        return await self._chat_completions_canonical(request)
 
     # -----------------------------------------------------------
     # Payload helpers
@@ -1332,7 +1365,10 @@ async def stream_completion(
 
         # Check for errors before streaming
         if response.status_code >= 400:
-            from src.core.common.exceptions import BackendError
+            status_code = response.status_code
+            rate_limit_details, retry_after_seconds = (
+                _retry_after_metadata_from_httpx_headers(response.headers)
+            )
 
             try:
                 # Read only first 10MB of error body to prevent DoS (consistent with other middleware)
@@ -1356,14 +1392,30 @@ async def stream_completion(
 
                 body_text = body_bytes.decode("utf-8", errors="ignore")
 
-                if logger.isEnabledFor(logging.ERROR):
-                    # Note: stream_completion doesn't have context access (protocol method)
-                    # Context correlation would require protocol change
+                # Operational HTTP errors: never use exc_info=True here — under concurrent
+                # asyncio work, sys.exc_info() can belong to another task and produces a
+                # misleading traceback on this log line.
+                preview = (
+                    (body_text[:500] + "...") if len(body_text) > 500 else body_text
+                )
+                if status_code == 429:
+                    if logger.isEnabledFor(logging.WARNING):
+                        logger.warning(
+                            "Anthropic API rate limited (HTTP 429): %s",
+                            preview or "(empty body)",
+                        )
+                elif 400 <= status_code < 500:
+                    if logger.isEnabledFor(logging.WARNING):
+                        logger.warning(
+                            "Anthropic API client error %s: %s",
+                            status_code,
+                            preview or "(empty body)",
+                        )
+                elif logger.isEnabledFor(logging.ERROR):
                     logger.error(
-                        "Anthropic API error %s: %s",
-                        response.status_code,
-                        body_text,
-                        exc_info=True,
+                        "Anthropic API server error %s: %s",
+                        status_code,
+                        preview or "(empty body)",
                     )
             except (UnicodeDecodeError, httpx.ReadError) as e:
                 if logger.isEnabledFor(logging.WARNING):
@@ -1375,10 +1427,17 @@ async def stream_completion(
                 body_text = ""
             finally:
                 await response.aclose()
+
+            if status_code == 429:
+                raise RateLimitExceededError(
+                    message=body_text or "Anthropic rate limit exceeded",
+                    details=rate_limit_details,
+                    reset_at=retry_after_seconds,
+                )
             raise BackendError(
                 message=body_text,
                 code="anthropic_error",
-                status_code=response.status_code,
+                status_code=status_code,
             )
         # Stream SSE messages
         try:

src/core/services/composite_failure_recovery_bridge.py

@@ -175,11 +175,21 @@ def _build_weighted_retry_request(
                 remaining.append(item)
 
         if not remaining:
-            weighted["excluded_selectors"] = excluded
-            context.extensions[COMPOSITE_ROUTING_STATE_KEY] = cast(
-                JsonValue, self._serialize_weighted_state(weighted)
-            )
-            return None
+            # All branches were exhausted once; recycle candidates by keeping only
+            # the current failed selector excluded and retrying the rest.
+            excluded = [selected]
+            excluded_set = {selected}
+            remaining = [
+                item
+                for item in weighted["branches"]
+                if item["selector"] not in excluded_set
+            ]
+            if not remaining:
+                weighted["excluded_selectors"] = excluded
+                context.extensions[COMPOSITE_ROUTING_STATE_KEY] = cast(
+                    JsonValue, self._serialize_weighted_state(weighted)
+                )
+                return None
 
         if len(remaining) == 1:
             next_selector = remaining[0]["selector"]

src/core/services/streaming/error_mapping.py

@@ -178,9 +178,25 @@ def map_backend_error(
                 status_code=status_code,
             )
 
-        # Map BackendError with quota_exceeded code
+        # Map BackendError: promote bare HTTP 429 to RateLimitExceededError so early
+        # streaming prefetch (integrate_streaming_pipeline) and terminal error chunks
+        # agree with connectors that raise RateLimitExceededError directly.
         if isinstance(error, BackendError):
-            # Preserve the BackendError as-is, including code and status_code
+            if (
+                not isinstance(error, RateLimitExceededError)
+                and getattr(error, "status_code", None) == 429
+            ):
+                merged_details = dict(error.details or {})
+                if "provider" not in merged_details:
+                    merged_details["provider"] = provider
+                if stream_id and "stream_id" not in merged_details:
+                    merged_details["stream_id"] = stream_id
+                return RateLimitExceededError(
+                    message=error.message,
+                    details=merged_details,
+                    reset_at=getattr(error, "reset_at", None),
+                )
+            # Preserve other BackendError variants as-is, including code and status_code
             return error
 
         # Map httpx connection errors

tests/integration/core/services/test_weighted_routing_recycles_on_transient_400.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, cast
+from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -9,6 +9,17 @@
 from src.core.domain.chat import ChatMessage, ChatRequest
 from src.core.domain.request_context import RequestContext
 from src.core.domain.responses import ResponseEnvelope
+from src.core.interfaces.backend_model_resolver_interface import ResolvedTarget
+from src.core.interfaces.domain_entities_interface import ISession
+from src.core.services.backend_completion_flow.failure_recovery_executor import (
+    FailureRecoveryExecutor,
+)
+from src.core.services.backend_completion_flow.service import BackendCompletionFlow
+from src.core.services.backend_model_resolver import BackendModelResolver
+from src.core.services.composite_failure_recovery_bridge import (
+    CompositeFailureRecoveryBridge,
+)
+from src.core.services.weighted_branch_selector import WeightedBranchSelector
 
 
 def _new_request_context() -> RequestContext:
@@ -196,7 +207,7 @@ def synchronize_request_with_target(
             request=request,
             resolved=target,
         )
-        return cast(ChatRequest, synchronized)
+        return synchronized
 
     async def prepare_backend_request(
         self,
@@ -366,7 +377,10 @@ def _rng() -> float:
     state = context.extensions["composite_routing_state"]
     assert isinstance(state, dict)
     assert state["selected_selector"] == "qwen-oauth:coder-model"
-    assert "zai-coding-plan:glm-5.1" in state["excluded_selectors"]
+    excluded_raw = state["excluded_selectors"]
+    assert isinstance(excluded_raw, list)
+    assert all(isinstance(x, str) for x in excluded_raw)
+    assert "zai-coding-plan:glm-5.1" in excluded_raw
     assert state["hop_count"] == 1
 
 
@@ -437,7 +451,7 @@ def _rng() -> float:
     state = context.extensions["composite_routing_state"]
     assert isinstance(state, dict)
     assert state["selected_selector"] == "zai-coding-plan:glm-5.1"
-    assert state["hop_count"] == 3
+    assert state["hop_count"] == 2
 
 
 @pytest.mark.asyncio
@@ -508,4 +522,4 @@ def _rng() -> float:
     state = context.extensions["composite_routing_state"]
     assert isinstance(state, dict)
     assert state["selected_selector"] == "zai-coding-plan:glm-5.1"
-    assert state["hop_count"] == 3
+    assert state["hop_count"] == 2

tests/unit/connectors/test_anthropic_error_handling.py

@@ -4,6 +4,7 @@
 
 import httpx
 import pytest
+from src.core.domain.chat import CanonicalChatRequest, ChatMessage
 
 
 @pytest.mark.asyncio
@@ -119,34 +120,87 @@ async def mock_aiter_text():
         assert exc_info.value.code == "anthropic_error_rate_limit"
 
 
-@pytest.mark.asyncio
-async def test_zai_coding_plan_uses_openai_connector():
-    """Test that zai-coding-plan now inherits from OpenAI connector."""
-    from src.connectors.openai import OpenAIConnector
-    from src.connectors.zai_coding_plan import ZaiCodingPlanBackend
-
-    # Use minimal mock setup to avoid heavy initialization
-    client = MagicMock()
-    config = MagicMock()
-    translation_service = MagicMock()
-
-    backend = ZaiCodingPlanBackend(client, config, translation_service)
-
-    # Verify it's an OpenAI connector now
-    assert isinstance(backend, OpenAIConnector)
-
-    # Mock _refresh_available_models to avoid network call entirely
-    async def mock_refresh():
-        backend.available_models = ["glm-4.6", "claude-sonnet-4-20250514"]
-        backend._provider_models = {"glm-4.6", "claude-sonnet-4-20250514"}
-
-    # Patch _refresh_available_models and directly set attributes to avoid initialization overhead
-    with patch.object(backend, "_refresh_available_models", new=mock_refresh):
-        # Directly set attributes that would be set during initialize
-        backend.api_key = "test-zai-key"
-        backend.api_base_url = "https://api.z.ai/api/coding/paas/v4"
-        backend._max_tokens_limit = 200000
-        backend._default_max_tokens = 8192
-
-    # Verify OpenAI-style API URL
-    assert "api.z.ai/api/coding/paas/v4" in backend.api_base_url
+@pytest.mark.asyncio
+async def test_stream_completion_http_429_raises_rate_limit_exceeded() -> None:
+    """HTTP 429 before the SSE body must map to RateLimitExceededError for resilience."""
+    from src.connectors.anthropic import AnthropicBackend
+    from src.core.common.exceptions import RateLimitExceededError
+    from src.core.config.app_config import AppConfig
+    from src.core.services.translation_service import TranslationService
+
+    client = httpx.AsyncClient()
+    config = AppConfig()
+    translation_service = TranslationService()
+
+    backend = AnthropicBackend(client, config, translation_service)
+    await backend.initialize(
+        anthropic_api_base_url="https://api.anthropic.com/v1",
+        key_name="test_key",
+        api_key="test-api-key-123",
+    )
+
+    err_json = (
+        '{"type":"error","error":{"type":"SubscriptionUsageLimitError",'
+        '"message":"quota exceeded"}}'
+    )
+    mock_response = MagicMock()
+    mock_response.status_code = 429
+    mock_response.headers = httpx.Headers({"retry-after": "42"})
+
+    async def mock_aiter_bytes():
+        yield err_json.encode()
+
+    mock_response.aiter_bytes = mock_aiter_bytes
+    mock_response.aclose = AsyncMock()
+
+    req = CanonicalChatRequest(
+        model="claude-3-5-sonnet-20241022",
+        messages=[ChatMessage(role="user", content="hello")],
+        stream=True,
+    )
+
+    with (
+        patch.object(backend.client, "build_request", return_value=MagicMock()),
+        patch.object(backend, "_capture_http_client") as cap,
+    ):
+        cap.send = AsyncMock(return_value=mock_response)
+        with pytest.raises(RateLimitExceededError) as exc_info:
+            async for _ in backend.stream_completion(req):
+                pass
+
+    assert "quota exceeded" in str(exc_info.value).lower()
+    assert exc_info.value.details.get("headers", {}).get("retry-after") == "42"
+    assert getattr(exc_info.value, "reset_at", None) == 42
+
+
+@pytest.mark.asyncio
+async def test_zai_coding_plan_uses_openai_connector():
+    """Test that zai-coding-plan now inherits from OpenAI connector."""
+    from src.connectors.openai import OpenAIConnector
+    from src.connectors.zai_coding_plan import ZaiCodingPlanBackend
+
+    # Use minimal mock setup to avoid heavy initialization
+    client = MagicMock()
+    config = MagicMock()
+    translation_service = MagicMock()
+
+    backend = ZaiCodingPlanBackend(client, config, translation_service)
+
+    # Verify it's an OpenAI connector now
+    assert isinstance(backend, OpenAIConnector)
+
+    # Mock _refresh_available_models to avoid network call entirely
+    async def mock_refresh():
+        backend.available_models = ["glm-4.6", "claude-sonnet-4-20250514"]
+        backend._provider_models = {"glm-4.6", "claude-sonnet-4-20250514"}
+
+    # Patch _refresh_available_models and directly set attributes to avoid initialization overhead
+    with patch.object(backend, "_refresh_available_models", new=mock_refresh):
+        # Directly set attributes that would be set during initialize
+        backend.api_key = "test-zai-key"
+        backend.api_base_url = "https://api.z.ai/api/coding/paas/v4"
+        backend._max_tokens_limit = 200000
+        backend._default_max_tokens = 8192
+
+    # Verify OpenAI-style API URL
+    assert "api.z.ai/api/coding/paas/v4" in backend.api_base_url

…s/unit/core/domain/test_session_state.py …_session_state_weighted_first_request.pytests/unit/core/domain/test_session_state.py renamed to tests/unit/core/domain/test_session_state_weighted_first_request.py tests/unit/core/domain/test_session_state.py renamed to tests/unit/core/domain/test_session_state_weighted_first_request.py

@@ -200,6 +200,23 @@ async def test_error_chunk_includes_retryable_flag(self) -> None:
         # The retryable flag should be present
         assert "retryable" in chunk.metadata["error"]
 
+    def test_streaming_error_mapper_promotes_backend_error_429(self) -> None:
+        """Plain BackendError(429) should map like a native rate-limit error."""
+
+        mapped_error = StreamingErrorMapper.map_backend_error(
+            BackendError(
+                message="upstream throttled",
+                status_code=429,
+                details={"headers": {"retry-after": "33"}},
+            ),
+            "anthropic",
+            "s-1",
+        )
+
+        assert isinstance(mapped_error, RateLimitExceededError)
+        assert mapped_error.details.get("headers", {}).get("retry-after") == "33"
+        assert mapped_error.details.get("stream_id") == "s-1"
+
     def test_streaming_error_mapper_preserves_retry_after_headers(self) -> None:
         """HTTP 429 detail headers should survive streaming error mapping."""