feat: actionable error messages and retry-with-fix for fixable 400s

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.

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,37 +190,109 @@ 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)
             self._first_backend_response = copy.deepcopy(response)
         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",

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",
+]