fix: canonicalize routing error details and propagate streaming errors to failover

Mateusz · Mateusz · commit 505d0494935b · 2026-04-10T23:10:51.000+02:00
diff --git a/.gitignore b/.gitignore
@@ -133,3 +133,9 @@ leveldb_test/
 
 # Cache directory
 var/cache/
+
+# Local test configs
+config/configtest1.yaml
+
+# Debug transcripts
+dev/transcript.md
diff --git a/src/core/services/backend_completion_flow/service.py b/src/core/services/backend_completion_flow/service.py
@@ -232,6 +232,55 @@ def _normalize_backend_exception(
 
         return candidate
 
+    @staticmethod
+    def _exception_from_streaming_error_envelope(
+        result: StreamingResponseEnvelope,
+    ) -> Exception | None:
+        """Convert terminal streaming error envelopes back into domain errors."""
+        status_code = getattr(result, "status_code", None)
+        if not isinstance(status_code, int) or status_code < 400:
+            return None
+
+        metadata = getattr(result, "metadata", None)
+        error_message = f"Backend returned {status_code} error"
+        error_type = ""
+        error_code = ""
+        error_details: dict[str, Any] = {}
+
+        if isinstance(metadata, dict):
+            raw_message = metadata.get("error_message")
+            if isinstance(raw_message, str) and raw_message.strip():
+                error_message = raw_message
+
+            raw_type = metadata.get("error_type")
+            if isinstance(raw_type, str):
+                error_type = raw_type
+
+            raw_code = metadata.get("error_code")
+            if isinstance(raw_code, str):
+                error_code = raw_code
+
+            raw_details = metadata.get("error_details")
+            if isinstance(raw_details, dict):
+                error_details = dict(raw_details)
+
+        if status_code == 401:
+            return AuthenticationError(error_message)
+
+        if status_code == 404 or error_type == "RoutingError":
+            return RoutingError(
+                message=error_message,
+                details=error_details,
+                code=error_details.get("code") or error_code or "unknown_model",
+            )
+
+        return BackendError(
+            message=error_message,
+            status_code=status_code,
+            details=error_details,
+            code=error_code or None,
+        )
+
     async def _enforce_non_forwardable_content(
         self,
         session_id: str,
@@ -1086,6 +1135,12 @@ def cancel(self) -> None:
 
                 # Step 10: Handle streaming response (wire capture + session ID injection)
                 if isinstance(result, StreamingResponseEnvelope):
+                    if allow_failover:
+                        streaming_error = self._exception_from_streaming_error_envelope(
+                            result
+                        )
+                        if streaming_error is not None:
+                            raise streaming_error
                     return await self._handle_streaming_response(
                         result=result,
                         backend_type=backend_type,
diff --git a/src/core/services/request_processor_service.py b/src/core/services/request_processor_service.py
@@ -48,6 +48,29 @@
 logger = logging.getLogger(__name__)
 
 MAX_QUALITY_VERIFIER_TURN_STATES = 10_000
+_ROUTING_CODES_BY_STATUS: dict[int, tuple[str, str, bool]] = {
+    400: ("unsupported_on_instance", "availability", False),
+    403: ("policy_rejected", "policy", False),
+    404: ("unknown_model", "validation", False),
+    503: ("temporarily_unavailable", "availability", True),
+}
+
+
+def _canonicalize_routing_error_details(
+    status_code: int,
+    details: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Normalize routing metadata so status and canonical code do not disagree."""
+    normalized = dict(details) if isinstance(details, dict) else {}
+    canonical = _ROUTING_CODES_BY_STATUS.get(status_code)
+    if canonical is None:
+        return normalized
+
+    code, category, retryable = canonical
+    normalized["code"] = code
+    normalized["category"] = category
+    normalized["retryable"] = retryable
+    return normalized
 
 
 class RequestProcessor(IRequestProcessor):
@@ -832,10 +855,14 @@ def _prepare_quality_verifier_extensions_for_backend_call(
                         error_message or "Backend returned 401 error"
                     )
                 elif result.status_code == 404 or orig_type == "RoutingError":
+                    error_details = _canonicalize_routing_error_details(
+                        int(result.status_code),
+                        error_details,
+                    )
                     raise RoutingError(
                         message=error_message,
                         details=error_details,
-                        code=orig_code or "unknown_model",
+                        code=error_details.get("code") or orig_code or "unknown_model",
                     )
                 else:
                     raise BackendError(
@@ -975,10 +1002,16 @@ def _prepare_quality_verifier_extensions_for_backend_call(
                                 fallback_result.status_code == 404
                                 or fallback_type == "RoutingError"
                             ):
+                                error_details = _canonicalize_routing_error_details(
+                                    int(fallback_result.status_code),
+                                    error_details,
+                                )
                                 raise RoutingError(
                                     message=error_message,
                                     details=error_details,
-                                    code=fallback_code or "unknown_model",
+                                    code=error_details.get("code")
+                                    or fallback_code
+                                    or "unknown_model",
                                 )
                             else:
                                 raise BackendError(
diff --git a/tests/unit/core/services/test_backend_completion_flow_streaming_error_envelope.py b/tests/unit/core/services/test_backend_completion_flow_streaming_error_envelope.py
@@ -5,8 +5,9 @@
 
 import pytest
 from fastapi import HTTPException
+from src.core.common.exceptions import RoutingError
 from src.core.domain.chat import ChatMessage, ChatRequest
-from src.core.domain.responses import StreamingResponseEnvelope
+from src.core.domain.responses import ResponseEnvelope, StreamingResponseEnvelope
 from src.core.services.backend_completion_flow.service import BackendCompletionFlow
 
 
@@ -70,6 +71,89 @@ def _build_flow_with_erroring_backend() -> BackendCompletionFlow:
     return BackendCompletionFlow(**deps)
 
 
+def _build_flow_with_streaming_error_envelope() -> (
+    tuple[BackendCompletionFlow, dict[str, Any]]
+):
+    deps: dict[str, Any] = {
+        "availability_checker": MagicMock(),
+        "request_preparer": MagicMock(),
+        "session_resolver": MagicMock(),
+        "backend_invoker": MagicMock(),
+        "failover_executor": MagicMock(),
+        "wire_capture_orchestrator": MagicMock(),
+        "usage_accounting_orchestrator": MagicMock(),
+        "exception_normalizer": MagicMock(),
+        "stream_formatting_service": MagicMock(),
+        "connector_invoker": MagicMock(),
+    }
+
+    deps["exception_normalizer"].normalize.side_effect = lambda exc, _backend: exc
+    deps["request_preparer"].prepare_request = AsyncMock(
+        return_value=MagicMock(
+            backend="ollama",
+            model="minimax-m2.7-cloud",
+            uri_params={},
+        )
+    )
+    deps["request_preparer"].synchronize_request_with_target.side_effect = (
+        lambda req, _target: req
+    )
+    deps["request_preparer"].prepare_backend_request = AsyncMock(
+        side_effect=lambda request, *_args, **_kwargs: request
+    )
+    deps["request_preparer"].prepare_backend_kwargs = MagicMock(return_value={})
+
+    deps["failover_executor"].check_complex_failover = AsyncMock(return_value=False)
+    deps["failover_executor"].apply_failure_recovery = AsyncMock(
+        return_value=ResponseEnvelope(content={"ok": True}, status_code=200)
+    )
+
+    deps["availability_checker"].check_backend_availability = AsyncMock()
+    deps["session_resolver"].resolve_session = AsyncMock(
+        return_value=(MagicMock(), "session-1")
+    )
+    deps["backend_invoker"].acquire_backend = AsyncMock(return_value=MagicMock())
+
+    deps["wire_capture_orchestrator"].prepare_wire_capture_context = AsyncMock(
+        return_value=None
+    )
+    deps["wire_capture_orchestrator"].capture_wire_outbound = AsyncMock()
+    deps["wire_capture_orchestrator"].capture_inbound_response = AsyncMock()
+    deps["wire_capture_orchestrator"].detect_key_name.return_value = "test-key"
+
+    deps["usage_accounting_orchestrator"].calculate_and_record_usage = AsyncMock(
+        return_value=(0, None, None)
+    )
+    deps["usage_accounting_orchestrator"].wrap_response_for_usage = AsyncMock(
+        side_effect=lambda **kwargs: kwargs["result"]
+    )
+    deps["usage_accounting_orchestrator"].handle_backend_error = AsyncMock()
+
+    deps["connector_invoker"].invoke = AsyncMock(
+        return_value=StreamingResponseEnvelope(
+            content=async_iter(()),
+            status_code=404,
+            metadata={
+                "error_message": "Backend returned 404 error",
+                "error_type": "RoutingError",
+                "error_code": "unknown_model",
+                "error_details": {
+                    "code": "unknown_model",
+                    "category": "validation",
+                    "retryable": False,
+                },
+            },
+        )
+    )
+
+    return BackendCompletionFlow(**deps), deps
+
+
+async def async_iter(items):
+    for item in items:
+        yield item
+
+
 @pytest.mark.asyncio
 async def test_streaming_call_returns_terminal_error_envelope_on_http_exception() -> (
     None
@@ -115,3 +199,34 @@ async def test_streaming_call_returns_terminal_error_envelope_on_http_exception(
     error_payload = metadata.get("error")
     assert isinstance(error_payload, dict)
     assert error_payload.get("status_code") == 413
+
+
+@pytest.mark.asyncio
+async def test_streaming_error_envelope_enters_failover_recovery_when_enabled() -> None:
+    flow, deps = _build_flow_with_streaming_error_envelope()
+
+    request = ChatRequest(
+        model="alias:minimax",
+        messages=[ChatMessage(role="user", content="continue")],
+    )
+
+    result = await flow.call_completion(
+        request=request,
+        stream=True,
+        allow_failover=True,
+        context=None,
+    )
+
+    assert isinstance(result, ResponseEnvelope)
+    assert result.content == {"ok": True}
+
+    deps["failover_executor"].apply_failure_recovery.assert_awaited_once()
+    recovery_error = deps["failover_executor"].apply_failure_recovery.await_args.kwargs[
+        "error"
+    ]
+    assert isinstance(recovery_error, RoutingError)
+    assert recovery_error.details == {
+        "code": "unknown_model",
+        "category": "validation",
+        "retryable": False,
+    }
diff --git a/tests/unit/core/services/test_request_processor_fallback.py b/tests/unit/core/services/test_request_processor_fallback.py
