diff --git a/changelog.d/actionable-errors-retry-fix.md b/changelog.d/actionable-errors-retry-fix.md new file mode 100644 index 000000000..b8488e1c1 --- /dev/null +++ b/changelog.d/actionable-errors-retry-fix.md @@ -0,0 +1,6 @@ +--- +category: Features +pr: 799 +--- + +**Actionable error messages and retry-with-fix for fixable 400s**: Errors returned to clients now append a human-readable `Suggestion:` line to the raw upstream message (non-streaming responses and mid-stream SSE error events), and the pipeline automatically retries once, with the offending field stripped, when the upstream API rejects a request with an "Extra inputs are not permitted" 400. Repairs are observable via a `pipeline.retry_with_fix` event and a warning log; the raw upstream error text is always preserved. diff --git a/src/luthien_proxy/pipeline/anthropic_processor.py b/src/luthien_proxy/pipeline/anthropic_processor.py index a479a7a21..d05a5a783 100644 --- a/src/luthien_proxy/pipeline/anthropic_processor.py +++ b/src/luthien_proxy/pipeline/anthropic_processor.py @@ -51,7 +51,15 @@ ) from luthien_proxy.observability.emitter import EventEmitterProtocol from luthien_proxy.pipeline.client_format import ClientFormat +from luthien_proxy.pipeline.error_advice import ( + CONNECTION_ERROR_ADVICE, + CREDENTIAL_ERROR_ADVICE, + INTERNAL_ERROR_ADVICE, + append_advice, + get_error_advice, +) from luthien_proxy.pipeline.policy_context_injection import inject_policy_awareness_anthropic +from luthien_proxy.pipeline.request_repair import attempt_request_fix from luthien_proxy.pipeline.session import ( extract_session_id_from_anthropic_body, extract_session_id_from_headers, @@ -182,14 +190,62 @@ def _record_backend_request(self, request: AnthropicRequest) -> None: endpoint="/v1/messages", ) + def _attempt_fixable_400_repair( + self, request: AnthropicRequest, error: AnthropicStatusError + ) -> AnthropicRequest | None: + """Return a repaired request for a known-fixable 400 error, or None. + + The repair is observable, never silent: a warning is logged and a + ``pipeline.retry_with_fix`` event is recorded (including the raw + upstream error and the removed field) before the caller retries. + Callers issue at most one retry per request; a failure of the + repaired request propagates normally. + """ + if (error.status_code or 0) != 400: + return None + fix = attempt_request_fix(request, str(error.message)) + if fix is None: + return None + logger.warning( + "[%s] Fixable 400 from upstream (%s); retrying once with repaired request", + self._call_id, + fix.description, + ) + self._emitter.record( + self._call_id, + "pipeline.retry_with_fix", + { + "original_error": str(error.message), + "removed_field": fix.removed_field, + "description": fix.description, + "session_id": self._session_id, + "user_id": self._user_id, + }, + ) + return fix.request + async def complete(self, request: AnthropicRequest | None = None) -> AnthropicResponse: - """Execute a non-streaming backend request.""" + """Execute a non-streaming backend request. + + Known-fixable 400 errors (e.g. an unrecognized extra field) trigger a + single retry with the repaired request; see _attempt_fixable_400_repair. + """ final_request = request or self._request self._record_backend_request(final_request) with tracer.start_as_current_span("send_upstream") as span: span.set_attribute("luthien.phase", "send_upstream") - response = await self._anthropic_client.complete(final_request, extra_headers=self._extra_headers) + try: + response = await self._anthropic_client.complete(final_request, extra_headers=self._extra_headers) + except AnthropicStatusError as e: + fixed_request = self._attempt_fixable_400_repair(final_request, e) + if fixed_request is None: + raise + # One retry max: the repaired call is not wrapped, so a second + # failure propagates to the normal error handling path. + self.set_request(fixed_request) + self._record_backend_request(fixed_request) + response = await self._anthropic_client.complete(fixed_request, extra_headers=self._extra_headers) if self._first_backend_response is None: # Deep-copy to preserve pre-policy content (policies may mutate in-place) @@ -197,22 +253,46 @@ async def complete(self, request: AnthropicRequest | None = None) -> AnthropicRe return response def stream(self, request: AnthropicRequest | None = None) -> AsyncIterator[MessageStreamEvent]: - """Execute a streaming backend request.""" + """Execute a streaming backend request. + + Known-fixable 400 errors trigger a single retry with the repaired + request, but only when the failure happens before any event has been + yielded (a later retry would duplicate events already delivered to + the policy/client). + """ final_request = request or self._request self._record_backend_request(final_request) extra_headers = self._extra_headers + async def _iterate(req: AnthropicRequest) -> AsyncIterator[MessageStreamEvent]: + async for event in self._anthropic_client.stream(req, extra_headers=extra_headers): + # RawMessageStreamEvent members are a subset of MessageStreamEvent; + # cast bridges Pyright's strict union checking. + mse = cast(MessageStreamEvent, event) + if self._buffer_raw_events: + self._raw_backend_events.append(mse) + yield mse + async def _stream() -> AsyncIterator[MessageStreamEvent]: with tracer.start_as_current_span("send_upstream") as span: span.set_attribute("luthien.phase", "send_upstream") - async for event in self._anthropic_client.stream(final_request, extra_headers=extra_headers): - # RawMessageStreamEvent members are a subset of MessageStreamEvent; - # cast bridges Pyright's strict union checking. - mse = cast(MessageStreamEvent, event) - if self._buffer_raw_events: - self._raw_backend_events.append(mse) - yield mse + yielded_any = False + try: + async for mse in _iterate(final_request): + yielded_any = True + yield mse + except AnthropicStatusError as e: + if yielded_any: + raise + fixed_request = self._attempt_fixable_400_repair(final_request, e) + if fixed_request is None: + raise + # One retry max: errors from the repaired stream propagate. + self.set_request(fixed_request) + self._record_backend_request(fixed_request) + async for mse in _iterate(fixed_request): + yield mse return _stream() @@ -1040,7 +1120,10 @@ async def _handle_execution_non_streaming( logger.error("[%s] Unexpected error in non-streaming policy execution: %s", call_id, e) raise BackendAPIError( status_code=500, - message=client_error_detail(str(e), "An internal error occurred while processing the request."), + message=append_advice( + client_error_detail(str(e), "An internal error occurred while processing the request."), + INTERNAL_ERROR_ADVICE, + ), error_type="api_error", client_format=ClientFormat.ANTHROPIC, ) from e @@ -1197,15 +1280,22 @@ def _build_error_event(e: Exception, call_id: str) -> _StreamErrorEvent: """ if isinstance(e, AnthropicStatusError): error_type = _ANTHROPIC_STATUS_ERROR_TYPE_MAP.get(e.status_code or 500, "api_error") - message = str(e.message) - logger.warning(f"[{call_id}] Mid-stream Anthropic API error: {e.status_code} {message}") + raw_message = str(e.message) + message = append_advice(raw_message, get_error_advice(e.status_code, raw_message)) + logger.warning(f"[{call_id}] Mid-stream Anthropic API error: {e.status_code} {raw_message}") elif isinstance(e, AnthropicConnectionError): error_type = "api_connection_error" - message = client_error_detail(str(e), "An error occurred while connecting to the API.") + message = append_advice( + client_error_detail(str(e), "An error occurred while connecting to the API."), + CONNECTION_ERROR_ADVICE, + ) logger.warning(f"[{call_id}] Mid-stream Anthropic connection error: {repr(e)}") else: error_type = "api_error" - message = client_error_detail(str(e), "An internal error occurred while processing the request.") + message = append_advice( + client_error_detail(str(e), "An internal error occurred while processing the request."), + INTERNAL_ERROR_ADVICE, + ) logger.error(f"[{call_id}] Mid-stream error: {repr(e)}") return _StreamErrorEvent( @@ -1250,9 +1340,12 @@ def _handle_anthropic_error(e: Exception, call_id: str) -> None: logger.warning(f"[{call_id}] Credential error during policy execution: {repr(e)}") raise BackendAPIError( status_code=502, - message=client_error_detail( - f"Credential resolution failed: {e}", - "The proxy could not authenticate to the backend service.", + message=append_advice( + client_error_detail( + f"Credential resolution failed: {e}", + "The proxy could not authenticate to the backend service.", + ), + CREDENTIAL_ERROR_ADVICE, ), error_type="credential_error", client_format=ClientFormat.ANTHROPIC, @@ -1262,9 +1355,10 @@ def _handle_anthropic_error(e: Exception, call_id: str) -> None: status_code = e.status_code or 500 error_type = _ANTHROPIC_STATUS_ERROR_TYPE_MAP.get(status_code, "api_error") logger.warning(f"[{call_id}] Anthropic API error: {status_code} {e.message}") + raw_message = str(e.message) raise BackendAPIError( status_code=status_code, - message=str(e.message), + message=append_advice(raw_message, get_error_advice(status_code, raw_message)), error_type=error_type, client_format=ClientFormat.ANTHROPIC, provider="anthropic", @@ -1273,7 +1367,10 @@ def _handle_anthropic_error(e: Exception, call_id: str) -> None: logger.warning(f"[{call_id}] Anthropic connection error: {repr(e)}") raise BackendAPIError( status_code=502, - message=client_error_detail(str(e), "An error occurred while connecting to the API."), + message=append_advice( + client_error_detail(str(e), "An error occurred while connecting to the API."), + CONNECTION_ERROR_ADVICE, + ), error_type="api_connection_error", client_format=ClientFormat.ANTHROPIC, provider="anthropic", diff --git a/src/luthien_proxy/pipeline/error_advice.py b/src/luthien_proxy/pipeline/error_advice.py new file mode 100644 index 000000000..0682a0f11 --- /dev/null +++ b/src/luthien_proxy/pipeline/error_advice.py @@ -0,0 +1,161 @@ +"""Actionable advice for errors surfaced to proxy clients. + +Raw upstream API errors are often opaque to end users (e.g. a bare +``"messages.0.bogus: Extra inputs are not permitted"``). This module maps +known error shapes to short, actionable suggestions that the pipeline +appends to the client-facing error message. + +Design invariant: the raw upstream message is always preserved. Advice is +appended after the original text, never substituted for it, so clients and +operators that rely on the exact upstream wording lose nothing. +""" + +from __future__ import annotations + +import re + +SUGGESTION_PREFIX = "Suggestion:" + +# Advice for errors that never come from the upstream API's HTTP layer. +CONNECTION_ERROR_ADVICE = ( + "The Luthien proxy could not reach the upstream API. Check the proxy host's " + "network connection and any custom base URL configuration, then retry." +) +CREDENTIAL_ERROR_ADVICE = ( + "The proxy's backend credentials could not be resolved. Check the API key or " + "OAuth token configured for the Luthien proxy, or contact your Luthien proxy administrator." +) +INTERNAL_ERROR_ADVICE = ( + "This error occurred inside the Luthien proxy, not the upstream API. Retry once; " + "if it persists, ask your Luthien proxy administrator to check the proxy logs." +) + +_GENERIC_ADVICE = ( + "Retry the request; if the error persists, ask your Luthien proxy administrator to check the proxy logs." +) + +_INVALID_REQUEST_ADVICE = ( + "The upstream API rejected this request as invalid. Fix the field named in the message above and resend." +) + +# Ordered rules: (status_code, message pattern or None, advice). +# First match wins. A None pattern is the fallback for that status code, +# so pattern-specific rules must come before their status fallback. +_ADVICE_RULES: tuple[tuple[int, re.Pattern[str] | None, str], ...] = ( + ( + 400, + re.compile(r"Extra inputs are not permitted", re.IGNORECASE), + "The API rejected a field it does not recognize (named just before " + "'Extra inputs are not permitted'). Remove that field from the request and resend.", + ), + ( + 400, + re.compile(r"max_tokens", re.IGNORECASE), + "Check the max_tokens value: it must be a positive integer within the " + "selected model's output limit. Lower it and resend.", + ), + ( + 400, + re.compile(r"credit balance", re.IGNORECASE), + "The Anthropic account behind this proxy has run out of credits. Add credits " + "in the Anthropic Console billing page, or contact your Luthien proxy administrator.", + ), + (400, None, _INVALID_REQUEST_ADVICE), + ( + 401, + None, + "The upstream API rejected the credentials. If you supply your own API key " + "through the proxy, verify it is valid and active. If the proxy operator manages " + "credentials, contact your Luthien proxy administrator.", + ), + ( + 403, + None, + "The credentials are valid but not allowed to perform this action. Confirm your " + "account or workspace has access to the requested model or feature.", + ), + ( + 404, + re.compile(r"model", re.IGNORECASE), + "The requested model was not found. Check the model name for typos and confirm your account has access to it.", + ), + ( + 404, + None, + "The requested resource was not found. Check the request path and any identifiers.", + ), + ( + 413, + None, + "The request payload is too large. Trim conversation history or large content blocks and resend.", + ), + (422, None, _INVALID_REQUEST_ADVICE), + ( + 429, + None, + "The upstream API rate limit was hit. Wait briefly and retry with backoff. If this " + "happens often, ask your Luthien proxy administrator about rate limits.", + ), + ( + 500, + None, + "The upstream API hit an internal error. This is usually transient: retry the " + "request, and check the provider's status page if it persists.", + ), + ( + 503, + None, + "The upstream API is temporarily unavailable. Retry with exponential backoff.", + ), + ( + 529, + None, + "The upstream API is temporarily overloaded. Retry with exponential backoff.", + ), +) + + +def get_error_advice(status_code: int | None, message: str) -> str: + """Return a short actionable suggestion for an upstream error. + + Args: + status_code: HTTP status code from the upstream API, if known. + message: Raw upstream error message (used for pattern-specific advice). + + Returns: + A human-readable suggestion. Falls back to generic retry guidance when + no specific rule matches, so callers can rely on always getting advice. + """ + for rule_status, pattern, advice in _ADVICE_RULES: + if status_code != rule_status: + continue + if pattern is None or pattern.search(message): + return advice + return _GENERIC_ADVICE + + +def append_advice(message: str, advice: str | None) -> str: + """Append a suggestion to an error message, preserving the original text. + + Returns the message unchanged when advice is None or empty, or when the + message already carries a suggestion (guards against double-appending if + an error is formatted twice on its way out). The double-append check + matches the exact join string this function produces, so an upstream + message that merely contains the word "Suggestion:" is not mistaken for + already-annotated output. + """ + if not advice: + return message + if f"\n\n{SUGGESTION_PREFIX} " in message: + return message + return f"{message}\n\n{SUGGESTION_PREFIX} {advice}" + + +__all__ = [ + "CONNECTION_ERROR_ADVICE", + "CREDENTIAL_ERROR_ADVICE", + "INTERNAL_ERROR_ADVICE", + "SUGGESTION_PREFIX", + "append_advice", + "get_error_advice", +] diff --git a/src/luthien_proxy/pipeline/request_repair.py b/src/luthien_proxy/pipeline/request_repair.py new file mode 100644 index 000000000..98a414bc2 --- /dev/null +++ b/src/luthien_proxy/pipeline/request_repair.py @@ -0,0 +1,100 @@ +"""Automatic repair of known-fixable 400 request errors. + +The Anthropic API rejects requests containing unrecognized fields with a +400 whose message names the offending field, e.g. +``"tools.0.bogus: Extra inputs are not permitted"``. When the field can be +located in the request payload, the pipeline strips it and retries the +backend call exactly once. + +Repairs are never silent: the caller (``_AnthropicPolicyIO`` in +``pipeline/anthropic_processor.py``) emits a ``pipeline.retry_with_fix`` +observability event and logs a warning before retrying. This module is pure: +it only computes the repaired request, it performs no I/O. +""" + +from __future__ import annotations + +import copy +import re +from dataclasses import dataclass +from typing import cast + +from luthien_proxy.llm.types.anthropic import AnthropicRequest + +# Matches ": Extra inputs are not permitted" anywhere in the +# upstream message. Path segments are dict keys or list indices separated by +# dots. The field path may be bare or wrapped in quotes/backticks. +_EXTRA_FIELD_PATTERN = re.compile( + r"(?:^|[\s'\"`(])([A-Za-z0-9_][A-Za-z0-9_.\-]*)['\"`]?: Extra inputs are not permitted" +) + +# Fields the pipeline itself relies on; never auto-remove these even if an +# upstream message implicates them (which would indicate a deeper problem +# that field-stripping cannot fix). +_PROTECTED_TOP_LEVEL_FIELDS = frozenset({"model", "messages", "max_tokens", "stream"}) + + +@dataclass(frozen=True) +class RequestFix: + """A repaired request plus a description of what changed. + + Attributes: + request: Deep copy of the original request with the fix applied. + removed_field: Dotted path of the field that was removed. + description: Human-readable summary of the repair (for logs/events). + """ + + request: AnthropicRequest + removed_field: str + description: str + + +def attempt_request_fix(request: AnthropicRequest, error_message: str) -> RequestFix | None: + """Try to repair a request rejected with a known-fixable 400 error. + + Currently handles one pattern: an unrecognized extra field + (": Extra inputs are not permitted"), which is removed from a deep + copy of the request. The original request is never mutated. + + Args: + request: The request payload that the upstream API rejected. + error_message: Raw upstream 400 error message. + + Returns: + A RequestFix when the offending field was located and removed, or + None when the error does not match a fixable pattern, the field path + cannot be resolved in the payload, or the field is load-bearing for + the pipeline (model, messages, max_tokens, stream). + """ + match = _EXTRA_FIELD_PATTERN.search(error_message) + if match is None: + return None + + field_path = match.group(1) + segments = field_path.split(".") + if len(segments) == 1 and segments[0] in _PROTECTED_TOP_LEVEL_FIELDS: + return None + + repaired: dict = copy.deepcopy(dict(request)) + container: object = repaired + for segment in segments[:-1]: + if isinstance(container, dict) and segment in container: + container = container[segment] + elif isinstance(container, list) and segment.isdigit() and int(segment) < len(container): + container = container[int(segment)] + else: + return None + + leaf = segments[-1] + if not isinstance(container, dict) or leaf not in container: + return None + del container[leaf] + + return RequestFix( + request=cast(AnthropicRequest, repaired), + removed_field=field_path, + description=f"removed field '{field_path}' rejected by the upstream API as an extra input", + ) + + +__all__ = ["RequestFix", "attempt_request_fix"] diff --git a/tests/luthien_proxy/unit_tests/pipeline/test_anthropic_processor.py b/tests/luthien_proxy/unit_tests/pipeline/test_anthropic_processor.py index 4df5a24bf..4b0443b0a 100644 --- a/tests/luthien_proxy/unit_tests/pipeline/test_anthropic_processor.py +++ b/tests/luthien_proxy/unit_tests/pipeline/test_anthropic_processor.py @@ -830,7 +830,10 @@ def test_builds_api_status_error_event(self): assert event.get("type") == "error" assert event.get("error", {}).get("type") == "rate_limit_error" - assert "Rate limit exceeded" in event.get("error", {}).get("message", "") + message = event.get("error", {}).get("message", "") + assert "Rate limit exceeded" in message + # Raw upstream error must be preserved AND an actionable suggestion appended. + assert "Suggestion:" in message def test_builds_connection_error_event(self): """Test building error event from AnthropicConnectionError.""" @@ -841,7 +844,9 @@ def test_builds_connection_error_event(self): assert event.get("type") == "error" assert event.get("error", {}).get("type") == "api_connection_error" - assert event.get("error", {}).get("message") == "An error occurred while connecting to the API." + message = event.get("error", {}).get("message", "") + assert "An error occurred while connecting to the API." in message + assert "Suggestion:" in message def test_builds_generic_error_event(self): """Generic exceptions produce a sanitized error event — internal details are not forwarded.""" @@ -851,7 +856,10 @@ def test_builds_generic_error_event(self): assert event.get("type") == "error" assert event.get("error", {}).get("type") == "api_error" - assert event.get("error", {}).get("message") == "An internal error occurred while processing the request." + message = event.get("error", {}).get("message", "") + assert "An internal error occurred while processing the request." in message + assert "Something went wrong" not in message # internal details stay sanitized + assert "Suggestion:" in message class TestMidStreamErrorHandling: @@ -1073,6 +1081,8 @@ def test_auth_error_raises_backend_api_error(self): assert exc_info.value.status_code == 401 assert exc_info.value.error_type == "authentication_error" assert "Invalid API Key" in exc_info.value.message + # Raw upstream message preserved AND actionable suggestion appended. + assert "Suggestion:" in exc_info.value.message def test_rate_limit_error_raises_backend_api_error(self): """429 RateLimitError should raise BackendAPIError with rate_limit_error type.""" @@ -1102,6 +1112,287 @@ def test_connection_error_raises_backend_api_error(self): assert exc_info.value.status_code == 502 assert exc_info.value.error_type == "api_connection_error" + assert "Suggestion:" in exc_info.value.message + + +def _fixable_400_error(message: str = "banana_mode: Extra inputs are not permitted") -> AnthropicStatusError: + """Build an AnthropicStatusError shaped like a fixable extra-field 400.""" + mock_response = HttpxResponse( + status_code=400, + request=HttpxRequest("POST", "https://api.anthropic.com/v1/messages"), + json={"error": {"type": "invalid_request_error", "message": message}}, + ) + return AnthropicStatusError( + message=message, + response=mock_response, + body={"error": {"type": "invalid_request_error", "message": message}}, + ) + + +class TestRetryWithFix: + """Retry-with-fix for known-fixable 400 errors (extra field stripping).""" + + @pytest.fixture + def mock_fastapi_request(self): + request = MagicMock() + request.headers = {} + request.method = "POST" + request.url = MagicMock() + request.url.path = "/v1/messages" + return request + + @pytest.fixture + def mock_anthropic_response(self) -> AnthropicResponse: + return AnthropicResponse( + id="msg_retry_ok", + type="message", + role="assistant", + content=[{"type": "text", "text": "Recovered!"}], + model=DEFAULT_TEST_MODEL, + stop_reason="end_turn", + stop_sequence=None, + usage={"input_tokens": 10, "output_tokens": 5}, + ) + + @pytest.mark.asyncio + async def test_non_streaming_fixable_400_retries_once_with_fix(self, mock_fastapi_request, mock_anthropic_response): + """A fixable 400 strips the offending field and retries exactly once.""" + anthropic_body: AnthropicRequest = { + "model": DEFAULT_TEST_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "max_tokens": 1024, + "stream": False, + "banana_mode": True, # type: ignore[typeddict-unknown-key] + } + mock_fastapi_request.json = AsyncMock(return_value=anthropic_body) + + mock_client = MagicMock() + mock_client.complete = AsyncMock(side_effect=[_fixable_400_error(), mock_anthropic_response]) + mock_emitter = MagicMock() + + with patch("luthien_proxy.pipeline.anthropic_processor.tracer") as mock_tracer: + mock_span = MagicMock() + mock_tracer.start_as_current_span.return_value.__enter__ = MagicMock(return_value=mock_span) + mock_tracer.start_as_current_span.return_value.__exit__ = MagicMock(return_value=False) + + response = await process_anthropic_request( + request=mock_fastapi_request, + policy=NoOpPolicy(), + anthropic_client=mock_client, + emitter=mock_emitter, + ) + + assert isinstance(response, JSONResponse) + payload = json.loads(bytes(response.body)) + assert payload["id"] == "msg_retry_ok" + + # Exactly two backend calls: original, then repaired. + assert mock_client.complete.call_count == 2 + retried_request = mock_client.complete.call_args_list[1][0][0] + assert "banana_mode" not in retried_request + assert retried_request["model"] == DEFAULT_TEST_MODEL + + # The repair is observable, not silent. + event_types = [call[0][1] for call in mock_emitter.record.call_args_list] + assert "pipeline.retry_with_fix" in event_types + for call in mock_emitter.record.call_args_list: + if call[0][1] == "pipeline.retry_with_fix": + event_payload = call[0][2] + assert event_payload["removed_field"] == "banana_mode" + assert "Extra inputs are not permitted" in event_payload["original_error"] + break + + # Audit trail shows BOTH backend attempts (original and repaired). + backend_requests = [ + call[0][2]["payload"] + for call in mock_emitter.record.call_args_list + if call[0][1] == "pipeline.backend_request" + ] + assert len(backend_requests) == 2 + assert "banana_mode" in backend_requests[0] + assert "banana_mode" not in backend_requests[1] + + @pytest.mark.asyncio + async def test_non_streaming_retry_capped_at_one_attempt(self, mock_fastapi_request): + """If the repaired request also fails, the error propagates: no retry loops.""" + anthropic_body: AnthropicRequest = { + "model": DEFAULT_TEST_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "max_tokens": 1024, + "stream": False, + "banana_mode": True, # type: ignore[typeddict-unknown-key] + } + mock_fastapi_request.json = AsyncMock(return_value=anthropic_body) + + mock_client = MagicMock() + mock_client.complete = AsyncMock(side_effect=[_fixable_400_error(), _fixable_400_error()]) + + with patch("luthien_proxy.pipeline.anthropic_processor.tracer") as mock_tracer: + mock_span = MagicMock() + mock_tracer.start_as_current_span.return_value.__enter__ = MagicMock(return_value=mock_span) + mock_tracer.start_as_current_span.return_value.__exit__ = MagicMock(return_value=False) + + with pytest.raises(BackendAPIError) as exc_info: + await process_anthropic_request( + request=mock_fastapi_request, + policy=NoOpPolicy(), + anthropic_client=mock_client, + emitter=MagicMock(), + ) + + assert mock_client.complete.call_count == 2 + assert exc_info.value.status_code == 400 + assert "Suggestion:" in exc_info.value.message + + @pytest.mark.asyncio + async def test_non_streaming_unfixable_400_is_not_retried(self, mock_fastapi_request): + """400s that don't match a fixable pattern propagate without a retry.""" + anthropic_body: AnthropicRequest = { + "model": DEFAULT_TEST_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "max_tokens": 1024, + "stream": False, + } + mock_fastapi_request.json = AsyncMock(return_value=anthropic_body) + + mock_client = MagicMock() + mock_client.complete = AsyncMock(side_effect=_fixable_400_error(message="messages: roles must alternate")) + + with patch("luthien_proxy.pipeline.anthropic_processor.tracer") as mock_tracer: + mock_span = MagicMock() + mock_tracer.start_as_current_span.return_value.__enter__ = MagicMock(return_value=mock_span) + mock_tracer.start_as_current_span.return_value.__exit__ = MagicMock(return_value=False) + + with pytest.raises(BackendAPIError) as exc_info: + await process_anthropic_request( + request=mock_fastapi_request, + policy=NoOpPolicy(), + anthropic_client=mock_client, + emitter=MagicMock(), + ) + + assert mock_client.complete.call_count == 1 + assert exc_info.value.status_code == 400 + assert "roles must alternate" in exc_info.value.message + assert "Suggestion:" in exc_info.value.message + + @pytest.mark.asyncio + async def test_streaming_fixable_400_before_first_event_retries_with_fix(self, mock_fastapi_request): + """A fixable 400 raised before any stream event triggers one repaired retry.""" + anthropic_body: AnthropicRequest = { + "model": DEFAULT_TEST_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "max_tokens": 1024, + "stream": True, + "banana_mode": True, # type: ignore[typeddict-unknown-key] + } + mock_fastapi_request.json = AsyncMock(return_value=anthropic_body) + + stream_requests: list[AnthropicRequest] = [] + + async def stream_fn(req, extra_headers=None): + stream_requests.append(req) + if len(stream_requests) == 1: + raise _fixable_400_error() + yield RawMessageStartEvent( + type="message_start", + message={ + "id": "msg_stream_retry", + "type": "message", + "role": "assistant", + "content": [], + "model": DEFAULT_TEST_MODEL, + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 0, "output_tokens": 0}, + }, + ) + yield RawMessageStopEvent(type="message_stop") + + mock_client = MagicMock() + mock_client.stream = stream_fn + mock_emitter = MagicMock() + + with patch("luthien_proxy.pipeline.anthropic_processor.tracer") as mock_tracer: + mock_span = MagicMock() + mock_tracer.start_as_current_span.return_value.__enter__ = MagicMock(return_value=mock_span) + mock_tracer.start_as_current_span.return_value.__exit__ = MagicMock(return_value=False) + + response = await process_anthropic_request( + request=mock_fastapi_request, + policy=NoOpPolicy(), + anthropic_client=mock_client, + emitter=mock_emitter, + ) + + chunks = [] + async for chunk in response.body_iterator: + chunks.append(chunk) + + combined = "".join(chunks) + assert "msg_stream_retry" in combined + assert "event: error" not in combined + + assert len(stream_requests) == 2 + assert "banana_mode" not in stream_requests[1] + + event_types = [call[0][1] for call in mock_emitter.record.call_args_list] + assert "pipeline.retry_with_fix" in event_types + + @pytest.mark.asyncio + async def test_streaming_fixable_400_after_events_is_not_retried(self, mock_fastapi_request): + """Once events have been yielded, a fixable 400 must NOT retry (would duplicate events).""" + anthropic_body: AnthropicRequest = { + "model": DEFAULT_TEST_MODEL, + "messages": [{"role": "user", "content": "Hi"}], + "max_tokens": 1024, + "stream": True, + "banana_mode": True, # type: ignore[typeddict-unknown-key] + } + mock_fastapi_request.json = AsyncMock(return_value=anthropic_body) + + stream_calls: list[AnthropicRequest] = [] + + async def stream_fn(req, extra_headers=None): + stream_calls.append(req) + yield RawMessageStartEvent( + type="message_start", + message={ + "id": "msg_partial", + "type": "message", + "role": "assistant", + "content": [], + "model": DEFAULT_TEST_MODEL, + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 0, "output_tokens": 0}, + }, + ) + raise _fixable_400_error() + + mock_client = MagicMock() + mock_client.stream = stream_fn + + with patch("luthien_proxy.pipeline.anthropic_processor.tracer") as mock_tracer: + mock_span = MagicMock() + mock_tracer.start_as_current_span.return_value.__enter__ = MagicMock(return_value=mock_span) + mock_tracer.start_as_current_span.return_value.__exit__ = MagicMock(return_value=False) + + response = await process_anthropic_request( + request=mock_fastapi_request, + policy=NoOpPolicy(), + anthropic_client=mock_client, + emitter=MagicMock(), + ) + + chunks = [] + async for chunk in response.body_iterator: + chunks.append(chunk) + + combined = "".join(chunks) + assert "event: error" in combined + assert "Suggestion:" in combined + assert len(stream_calls) == 1 class _InvalidStreamCompletePolicy: diff --git a/tests/luthien_proxy/unit_tests/pipeline/test_error_advice.py b/tests/luthien_proxy/unit_tests/pipeline/test_error_advice.py new file mode 100644 index 000000000..87f44c9d6 --- /dev/null +++ b/tests/luthien_proxy/unit_tests/pipeline/test_error_advice.py @@ -0,0 +1,83 @@ +"""Tests for actionable error advice (pipeline/error_advice.py).""" + +import pytest + +from luthien_proxy.pipeline.error_advice import ( + CONNECTION_ERROR_ADVICE, + CREDENTIAL_ERROR_ADVICE, + INTERNAL_ERROR_ADVICE, + SUGGESTION_PREFIX, + append_advice, + get_error_advice, +) + + +class TestGetErrorAdvice: + """Advice lookup for upstream status codes and message patterns.""" + + @pytest.mark.parametrize( + "status_code,message,expected_fragment", + [ + (400, "banana_mode: Extra inputs are not permitted", "does not recognize"), + (400, "max_tokens: 999999 > 64000, which is the maximum", "max_tokens"), + (400, "Your credit balance is too low to access the API", "credits"), + (400, "messages: roles must alternate", "rejected this request as invalid"), + (401, "invalid x-api-key", "credentials"), + (403, "forbidden", "access"), + (404, "model: claude-nonexistent not found", "model name"), + (404, "not found", "resource was not found"), + (413, "payload too large", "too large"), + (422, "invalid body", "rejected this request as invalid"), + (429, "rate limit exceeded", "retry with backoff"), + (500, "internal server error", "transient"), + (529, "overloaded", "overloaded"), + ], + ) + def test_known_errors_get_specific_advice(self, status_code, message, expected_fragment): + advice = get_error_advice(status_code, message) + assert expected_fragment in advice + + def test_unknown_status_gets_generic_advice(self): + advice = get_error_advice(418, "I'm a teapot") + assert advice + assert "Retry" in advice + + def test_none_status_gets_generic_advice(self): + advice = get_error_advice(None, "mystery error") + assert advice + + def test_none_status_and_empty_message_still_gets_advice(self): + """Callers can rely on always getting non-empty advice.""" + assert get_error_advice(None, "") + + +class TestAppendAdvice: + """Advice is appended without destroying the raw upstream message.""" + + def test_preserves_raw_message(self): + raw = "banana_mode: Extra inputs are not permitted" + combined = append_advice(raw, "Remove the field.") + assert combined.startswith(raw) + assert f"{SUGGESTION_PREFIX} Remove the field." in combined + + def test_none_advice_returns_message_unchanged(self): + assert append_advice("raw error", None) == "raw error" + + def test_empty_advice_returns_message_unchanged(self): + assert append_advice("raw error", "") == "raw error" + + def test_does_not_double_append(self): + once = append_advice("raw error", "Do the thing.") + twice = append_advice(once, "Do the other thing.") + assert twice == once + + def test_upstream_message_mentioning_suggestion_still_gets_advice(self): + """A raw message that merely contains 'Suggestion:' is not mistaken for annotated output.""" + raw = "field Suggestion: is not a valid tool name" + combined = append_advice(raw, "Rename the tool.") + assert combined.startswith(raw) + assert "Rename the tool." in combined + + def test_module_advice_constants_are_actionable(self): + for advice in (CONNECTION_ERROR_ADVICE, CREDENTIAL_ERROR_ADVICE, INTERNAL_ERROR_ADVICE): + assert "Luthien proxy" in advice diff --git a/tests/luthien_proxy/unit_tests/pipeline/test_request_repair.py b/tests/luthien_proxy/unit_tests/pipeline/test_request_repair.py new file mode 100644 index 000000000..5a05778b2 --- /dev/null +++ b/tests/luthien_proxy/unit_tests/pipeline/test_request_repair.py @@ -0,0 +1,99 @@ +"""Tests for fixable-400 request repair (pipeline/request_repair.py).""" + +import copy + +from luthien_proxy.llm.types.anthropic import AnthropicRequest +from luthien_proxy.pipeline.request_repair import attempt_request_fix + + +def _base_request() -> AnthropicRequest: + return { + "model": "claude-sonnet-4-5", + "messages": [{"role": "user", "content": "Hi"}], + "max_tokens": 1024, + } + + +class TestAttemptRequestFix: + """Repair of the 'Extra inputs are not permitted' 400 pattern.""" + + def test_strips_top_level_extra_field(self): + request = _base_request() + request["banana_mode"] = True # type: ignore[typeddict-unknown-key] + + fix = attempt_request_fix(request, "banana_mode: Extra inputs are not permitted") + + assert fix is not None + assert fix.removed_field == "banana_mode" + assert "banana_mode" not in fix.request + assert fix.request["model"] == request["model"] + assert "banana_mode" in fix.description + + def test_strips_nested_field_through_list_index(self): + request = _base_request() + request["messages"] = [{"role": "user", "content": "Hi", "bogus": 1}] + + fix = attempt_request_fix(request, "messages.0.bogus: Extra inputs are not permitted") + + assert fix is not None + assert fix.removed_field == "messages.0.bogus" + assert "bogus" not in fix.request["messages"][0] + assert fix.request["messages"][0]["role"] == "user" + + def test_original_request_is_not_mutated(self): + request = _base_request() + request["banana_mode"] = True # type: ignore[typeddict-unknown-key] + snapshot = copy.deepcopy(dict(request)) + + attempt_request_fix(request, "banana_mode: Extra inputs are not permitted") + + assert dict(request) == snapshot + + def test_protected_field_is_not_removed(self): + request = _base_request() + + fix = attempt_request_fix(request, "max_tokens: Extra inputs are not permitted") + + assert fix is None + + def test_non_matching_message_returns_none(self): + request = _base_request() + + fix = attempt_request_fix(request, "messages: roles must alternate") + + assert fix is None + + def test_field_absent_from_payload_returns_none(self): + request = _base_request() + + fix = attempt_request_fix(request, "ghost_field: Extra inputs are not permitted") + + assert fix is None + + def test_unresolvable_nested_path_returns_none(self): + request = _base_request() + + fix = attempt_request_fix(request, "messages.9.bogus: Extra inputs are not permitted") + + assert fix is None + + def test_field_path_json_quoted_in_message(self): + request = _base_request() + request["banana_mode"] = True # type: ignore[typeddict-unknown-key] + + fix = attempt_request_fix(request, '"banana_mode": Extra inputs are not permitted') + + assert fix is not None + assert fix.removed_field == "banana_mode" + + def test_field_path_embedded_in_longer_message(self): + request = _base_request() + request["banana_mode"] = True # type: ignore[typeddict-unknown-key] + + fix = attempt_request_fix( + request, + 'Error code: 400 - {"error": {"message": "banana_mode: Extra inputs are not permitted"}}', + ) + + assert fix is not None + assert fix.removed_field == "banana_mode"