feat: opt-in passthrough fallback when a policy modification causes an upstream 4xx

.env.example

@@ -69,6 +69,10 @@
 # Max rows per policy namespace in PolicyCache (0 or negative disables the cap)
 # POLICY_CACHE_MAX_ENTRIES=10000
 
+# When a policy-modified request fails upstream with a request-shaped 4xx (400/404/413/422), retry once with the original unmodified request. Fires only if the policy actually changed the request; streaming falls back only before any backend event arrived. Emits a pipeline.passthrough_fallback event when it fires. Off by default: the retry bypasses request-side policy modifications (fail-open), which weakens policies that rewrite requests for safety
+# (can also be set at runtime via admin API)
+# PASSTHROUGH_FALLBACK_ENABLED=false
+
 
 # === DATABASE ====================================================
 

changelog.d/passthrough-fallback.md

@@ -0,0 +1,9 @@
+---
+category: Features
+pr: 797
+---
+
+**Opt-in passthrough fallback** (`PASSTHROUGH_FALLBACK_ENABLED`, default off): when a policy-modified request is rejected upstream with a request-shaped 4xx (400/404/413/422), the gateway retries once with the original unmodified request so the proxy is never worse than direct API access
+  - Fires only when the policy actually changed the request; streaming falls back only before any backend event arrived
+  - Observable: emits a `pipeline.passthrough_fallback` event and a WARNING log when it fires — policy failures are never silently masked
+  - Intentional policy blocks are unaffected: blocks are policy-layer decisions and never surface as upstream errors, so the fallback structurally cannot override them

src/luthien_proxy/config_fields.py

@@ -146,6 +146,11 @@ class ConfigFieldMeta:
         "Max rows per policy namespace in PolicyCache (0 or negative disables the cap)",
         category="policy",
     ),
+    ConfigFieldMeta(
+        "passthrough_fallback_enabled", "PASSTHROUGH_FALLBACK_ENABLED", bool, False,
+        "When a policy-modified request fails upstream with a request-shaped 4xx (400/404/413/422), retry once with the original unmodified request. Fires only if the policy actually changed the request; streaming falls back only before any backend event arrived. Emits a pipeline.passthrough_fallback event when it fires. Off by default: the retry bypasses request-side policy modifications (fail-open), which weakens policies that rewrite requests for safety",
+        category="policy", db_settable=True, restart_required=False,
+    ),
 
     # ── database ──────────────────────────────────────────────────────────
     ConfigFieldMeta(

src/luthien_proxy/pipeline/anthropic_processor.py

@@ -96,6 +96,17 @@ class _StreamErrorEvent(TypedDict):
 tracer = trace.get_tracer(__name__)
 
 
+# Upstream status codes eligible for passthrough fallback: failures plausibly
+# caused by the *content* of the request body (which a policy modification can
+# break). Deliberately excludes:
+#   401/403 — credential/permission scoped; re-sending a different body with
+#             the same credential won't change the outcome,
+#   429     — rate limited; an immediate retry amplifies load and the original
+#             request would be throttled identically,
+#   5xx/529 — server-side; the Anthropic SDK already retries these itself.
+_PASSTHROUGH_FALLBACK_STATUS_CODES: frozenset[int] = frozenset({400, 404, 413, 422})
+
+
 class _AnthropicPolicyIO(AnthropicPolicyIOProtocol):
     """Request-scoped I/O helpers for execution-oriented Anthropic policies."""
 
@@ -111,6 +122,7 @@ def __init__(
         request_log_recorder: RequestLogRecorder,
         is_streaming: bool,
         extra_headers: dict[str, str] | None = None,
+        passthrough_fallback_enabled: bool = False,
     ) -> None:
         self._request = initial_request
         self._initial_request = initial_request
@@ -124,6 +136,14 @@ def __init__(
         self._extra_headers = extra_headers
         self._request_recorded = False
         self._first_backend_response: AnthropicResponse | None = None
+        # Pristine snapshot of the request as it entered the policy, used by
+        # passthrough fallback. deepcopy (not dict()) because policies may
+        # mutate nested message structures in place, which would corrupt a
+        # shallow copy. Taken only when the feature is enabled so the default
+        # path stays copy-free (no-op stays no-op).
+        self._fallback_original_request: AnthropicRequest | None = (
+            copy.deepcopy(initial_request) if passthrough_fallback_enabled else None
+        )
         # Raw backend events are only buffered when needed for non-streaming
         # response reconstruction (e.g., diff recording). Streaming responses
         # can reconstruct from the post-policy accumulated_events instead,
@@ -182,14 +202,75 @@ def _record_backend_request(self, request: AnthropicRequest) -> None:
             endpoint="/v1/messages",
         )
 
+    def _passthrough_fallback_request(
+        self, sent_request: AnthropicRequest, exc: AnthropicStatusError
+    ) -> AnthropicRequest | None:
+        """Return the original request to retry with, or None if fallback doesn't apply.
+
+        Fallback applies only when ALL of:
+        - the feature is enabled (PASSTHROUGH_FALLBACK_ENABLED),
+        - the upstream failure is request-shaped (400/404/413/422), and
+        - the policy actually changed the request — if the request is
+          byte-identical to what entered the policy, direct API access would
+          have failed identically and a retry is pure waste.
+
+        This deliberately lives at the backend-call site: intentional policy
+        blocks (response rewrites, synthetic block messages, policy-raised
+        errors) never surface as an upstream AnthropicStatusError from this
+        call, so fallback structurally cannot override a block.
+        """
+        original = self._fallback_original_request
+        if original is None:  # feature disabled
+            return None
+        if exc.status_code not in _PASSTHROUGH_FALLBACK_STATUS_CODES:
+            return None
+        if sent_request == original:
+            return None
+        return original
+
+    def _record_passthrough_fallback(self, exc: AnthropicStatusError) -> None:
+        """Make the fallback observable: WARNING log + pipeline event.
+
+        Recorded BEFORE the retry is attempted so the policy failure is never
+        silently masked, even if the retry itself then succeeds or fails.
+        """
+        logger.warning(
+            "[%s] Policy-modified request rejected upstream (%s: %s); falling back to the original unmodified request",
+            self._call_id,
+            exc.status_code,
+            exc.message,
+        )
+        self._emitter.record(
+            self._call_id,
+            "pipeline.passthrough_fallback",
+            {
+                "summary": "Policy-modified request failed upstream; retrying with original unmodified request",
+                "status_code": exc.status_code,
+                "error_message": str(exc.message),
+                "session_id": self._session_id,
+                "user_id": self._user_id,
+            },
+        )
+
     async def complete(self, request: AnthropicRequest | None = None) -> AnthropicResponse:
         """Execute a non-streaming backend request."""
         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 exc:
+                fallback_request = self._passthrough_fallback_request(final_request, exc)
+                if fallback_request is None:
+                    raise
+                self._record_passthrough_fallback(exc)
+                span.set_attribute("luthien.passthrough_fallback", True)
+                self._record_backend_request(fallback_request)
+                # If this retry also fails, the error propagates normally —
+                # the client sees exactly what direct API access would return.
+                response = await self._anthropic_client.complete(fallback_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)
@@ -203,16 +284,41 @@ def stream(self, request: AnthropicRequest | None = None) -> AsyncIterator[Messa
 
         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
+                events_yielded = 0
+                try:
+                    async for mse in _iterate(final_request):
+                        events_yielded += 1
+                        yield mse
+                except AnthropicStatusError as exc:
+                    # Fallback only if the failure happened at stream connect.
+                    # After events have flowed, the policy (and possibly the
+                    # client) already consumed part of the stream — re-sending
+                    # would duplicate or interleave content. No mid-stream
+                    # recovery, matching the PR #204 design discussion.
+                    if events_yielded:
+                        raise
+                    fallback_request = self._passthrough_fallback_request(final_request, exc)
+                    if fallback_request is None:
+                        raise
+                    self._record_passthrough_fallback(exc)
+                    span.set_attribute("luthien.passthrough_fallback", True)
+                    self._record_backend_request(fallback_request)
+                    # If this retry also fails, the error propagates normally —
+                    # the client sees exactly what direct API access would return.
+                    async for mse in _iterate(fallback_request):
+                        yield mse
 
         return _stream()
 
@@ -635,6 +741,7 @@ async def _execute_anthropic_policy(
         request_log_recorder=request_log_recorder,
         is_streaming=is_streaming,
         extra_headers=extra_headers,
+        passthrough_fallback_enabled=get_settings().passthrough_fallback_enabled,
     )
     emissions = _run_policy_hooks(execution_policy, io, policy_ctx)
 

src/luthien_proxy/settings.py

@@ -64,6 +64,7 @@ class Settings(_SettingsBase):
     inject_policy_context: bool = True
     dogfood_mode: bool = False
     policy_cache_max_entries: int = 10000
+    passthrough_fallback_enabled: bool = False
 
     # ── database ────────────────────────────────────────────────────
     database_url: str = ""