From 7a9a07e7fd02388e6dad24410ab0a4a715146c18 Mon Sep 17 00:00:00 2001 From: Scott Wofford Date: Mon, 6 Jul 2026 22:45:07 -0700 Subject: [PATCH 1/2] chore: set objective to implement opt-in passthrough fallback for policy-caused upstream failures From ec3fab52b11105e3de0975a98138c0a1c7b0d942 Mon Sep 17 00:00:00 2001 From: Scott Wofford Date: Mon, 6 Jul 2026 22:55:19 -0700 Subject: [PATCH 2/2] feat: opt-in passthrough fallback when a policy modification causes an upstream 4xx MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements https://trello.com/c/kRPRjGUx (COE audit follow-up to PR #204). Design principle: the proxy should never make things worse than direct API access. Trigger (all must hold): - PASSTHROUGH_FALLBACK_ENABLED is on (default off) - upstream AnthropicStatusError with status in {400, 404, 413, 422} - the request sent differs from the pristine pre-policy snapshot - (streaming) zero backend events received yet The fallback lives at the backend-call site (_AnthropicPolicyIO), so intentional policy blocks — which are response rewrites or policy-raised errors, never upstream errors — structurally cannot be overridden. Co-Authored-By: Claude Fable 5 --- .env.example | 4 + changelog.d/passthrough-fallback.md | 9 + src/luthien_proxy/config_fields.py | 5 + .../pipeline/anthropic_processor.py | 123 +++++- src/luthien_proxy/settings.py | 1 + .../test_mock_passthrough_fallback.py | 204 ++++++++++ .../pipeline/test_anthropic_processor.py | 350 ++++++++++++++++++ 7 files changed, 688 insertions(+), 8 deletions(-) create mode 100644 changelog.d/passthrough-fallback.md create mode 100644 tests/luthien_proxy/e2e_tests/test_mock_passthrough_fallback.py diff --git a/.env.example b/.env.example index 5b8ce8a8d..41425a18c 100644 --- a/.env.example +++ b/.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 ==================================================== diff --git a/changelog.d/passthrough-fallback.md b/changelog.d/passthrough-fallback.md new file mode 100644 index 000000000..f5e1fa1f6 --- /dev/null +++ b/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 diff --git a/src/luthien_proxy/config_fields.py b/src/luthien_proxy/config_fields.py index ed08ef169..663fc25a0 100644 --- a/src/luthien_proxy/config_fields.py +++ b/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( diff --git a/src/luthien_proxy/pipeline/anthropic_processor.py b/src/luthien_proxy/pipeline/anthropic_processor.py index a479a7a21..8e74f33bc 100644 --- a/src/luthien_proxy/pipeline/anthropic_processor.py +++ b/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,6 +202,56 @@ 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 @@ -189,7 +259,18 @@ async def complete(self, request: AnthropicRequest | None = None) -> AnthropicRe 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) diff --git a/src/luthien_proxy/settings.py b/src/luthien_proxy/settings.py index d361a02a5..4d3abf936 100644 --- a/src/luthien_proxy/settings.py +++ b/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 = "" diff --git a/tests/luthien_proxy/e2e_tests/test_mock_passthrough_fallback.py b/tests/luthien_proxy/e2e_tests/test_mock_passthrough_fallback.py new file mode 100644 index 000000000..0dcd03926 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/test_mock_passthrough_fallback.py @@ -0,0 +1,204 @@ +"""Mock e2e tests for the opt-in passthrough fallback (PASSTHROUGH_FALLBACK_ENABLED). + +Design principle (Trello kRPRjGUx, PR #204 follow-up): the proxy should never +make things worse than direct API access. When a policy-modified request is +rejected upstream with a request-shaped 4xx, the gateway retries once with the +original unmodified request — observably (pipeline.passthrough_fallback event ++ WARNING log), and only when the policy actually changed the request. + +The feature is OFF by default: the retry bypasses request-side policy +modifications (fail-open), which weakens policies that rewrite requests for +safety. These tests enable it via the admin config API and restore afterwards. + +400 errors are NOT retried by the Anthropic SDK, so a single enqueued error +maps to exactly one gateway-visible failure (no retry-slot bookkeeping needed). + +Run: + ./scripts/run_e2e.sh mock + # or directly: + uv run pytest -m mock_e2e tests/luthien_proxy/e2e_tests/test_mock_passthrough_fallback.py -v +""" + +import json +from contextlib import asynccontextmanager + +import httpx +import pytest +from tests.luthien_proxy.e2e_tests.conftest import policy_context +from tests.luthien_proxy.e2e_tests.mock_anthropic.responses import error_response, text_response +from tests.luthien_proxy.e2e_tests.mock_anthropic.server import MockAnthropicServer + +pytestmark = pytest.mark.mock_e2e + +# StringReplacementPolicy with apply_to="request" rewrites "hello" in the +# client request before it reaches the backend — a real request-modifying +# policy, so the fallback path is exercised end to end. +_MODIFYING_POLICY_REF = "luthien_proxy.policies.string_replacement_policy:StringReplacementPolicy" +_MODIFYING_POLICY_CONFIG = { + "replacements": [["hello", "POLICY-REWRITTEN"]], + "apply_to": "request", +} + +_BASE_REQUEST = { + "model": "claude-haiku-4-5", + "messages": [{"role": "user", "content": "hello from the client"}], + "max_tokens": 100, +} + + +@asynccontextmanager +async def _passthrough_fallback_enabled(gateway_url: str, admin_api_key: str): + """Enable PASSTHROUGH_FALLBACK_ENABLED via the admin config API; restore after. + + The flag defaults to False with source=default, so restoring = deleting the + DB override. Skips the test if the flag is pinned by env/CLI (409). + """ + config_url = f"{gateway_url}/api/admin/config/passthrough_fallback_enabled" + headers = {"Authorization": f"Bearer {admin_api_key}"} + async with httpx.AsyncClient(timeout=10.0) as client: + enable = await client.put(config_url, headers=headers, json={"value": True}) + if enable.status_code == 409: + pytest.skip( + "passthrough_fallback_enabled is overridden by env or CLI — cannot toggle via DB. " + "Unset PASSTHROUGH_FALLBACK_ENABLED in the gateway's environment to run this test." + ) + assert enable.status_code == 200, f"Failed to enable passthrough fallback: {enable.text}" + try: + yield + finally: + restore = await client.delete(config_url, headers=headers) + assert restore.status_code == 200, f"Failed to restore passthrough fallback config: {restore.text}" + + +def _message_content(request_body: dict) -> str: + return request_body["messages"][0]["content"] + + +@pytest.mark.asyncio +async def test_fallback_forwards_original_request_when_modified_request_400s( + mock_anthropic: MockAnthropicServer, + gateway_healthy, + gateway_url, + auth_headers, + admin_api_key, +): + """Policy modification causes a 400 -> the gateway retries with the + original unmodified request and the client gets the successful response.""" + mock_anthropic.enqueue(error_response(400, "invalid_request_error", "modified request rejected")) + mock_anthropic.enqueue(text_response("fallback succeeded")) + + async with _passthrough_fallback_enabled(gateway_url, admin_api_key): + async with policy_context( + _MODIFYING_POLICY_REF, + _MODIFYING_POLICY_CONFIG, + gateway_url=gateway_url, + admin_api_key=admin_api_key, + ): + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post( + f"{gateway_url}/v1/messages", + json={**_BASE_REQUEST, "stream": False}, + headers=auth_headers, + ) + + assert response.status_code == 200, f"Expected 200 after fallback, got {response.status_code}: {response.text}" + body = response.json() + assert body["content"][0]["text"] == "fallback succeeded" + + # The backend saw exactly two requests: the policy-modified one, then the + # original unmodified one. + requests_seen = mock_anthropic.received_requests() + assert len(requests_seen) == 2, f"Expected 2 backend requests, got {len(requests_seen)}" + # endswith: the gateway may prefix the first user message with the + # injection (INJECT_POLICY_CONTEXT defaults to true); + # the fallback restores the request as it entered the POLICY, so the + # injection prefix is present on both attempts. + assert _message_content(requests_seen[0]).endswith("POLICY-REWRITTEN from the client") + assert "hello" not in _message_content(requests_seen[0]) + assert _message_content(requests_seen[1]).endswith("hello from the client") + + +@pytest.mark.asyncio +async def test_fallback_disabled_by_default_propagates_error( + mock_anthropic: MockAnthropicServer, + gateway_healthy, + gateway_url, + auth_headers, + admin_api_key, +): + """With the flag at its default (off), the 400 from the policy-modified + request propagates to the client and no retry is attempted.""" + mock_anthropic.enqueue(error_response(400, "invalid_request_error", "modified request rejected")) + + async with policy_context( + _MODIFYING_POLICY_REF, + _MODIFYING_POLICY_CONFIG, + gateway_url=gateway_url, + admin_api_key=admin_api_key, + ): + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post( + f"{gateway_url}/v1/messages", + json={**_BASE_REQUEST, "stream": False}, + headers=auth_headers, + ) + + assert response.status_code == 400, f"Expected 400 with fallback off, got {response.status_code}" + body = response.json() + assert body.get("type") == "error" + assert body["error"]["type"] == "invalid_request_error" + assert len(mock_anthropic.received_requests()) == 1, "No retry should happen with fallback disabled" + + +@pytest.mark.asyncio +async def test_streaming_fallback_streams_original_request( + mock_anthropic: MockAnthropicServer, + gateway_healthy, + gateway_url, + auth_headers, + admin_api_key, +): + """Streaming: a 400 at stream connect falls back to streaming the original + unmodified request; the client receives a normal SSE stream.""" + mock_anthropic.enqueue(error_response(400, "invalid_request_error", "modified request rejected")) + mock_anthropic.enqueue(text_response("streamed fallback")) + + async with _passthrough_fallback_enabled(gateway_url, admin_api_key): + async with policy_context( + _MODIFYING_POLICY_REF, + _MODIFYING_POLICY_CONFIG, + gateway_url=gateway_url, + admin_api_key=admin_api_key, + ): + async with httpx.AsyncClient(timeout=30.0) as client: + async with client.stream( + "POST", + f"{gateway_url}/v1/messages", + json={**_BASE_REQUEST, "stream": True}, + headers=auth_headers, + ) as response: + assert response.status_code == 200 + raw_sse = "" + async for chunk in response.aiter_text(): + raw_sse += chunk + + # The stream carries the fallback response text and no error event. + text_parts: list[str] = [] + for line in raw_sse.splitlines(): + if not line.startswith("data: "): + continue + data = json.loads(line[len("data: ") :]) + assert data.get("type") != "error", f"Unexpected error event in fallback stream: {data}" + if data.get("type") == "content_block_delta" and data["delta"].get("type") == "text_delta": + text_parts.append(data["delta"]["text"]) + assert "".join(text_parts) == "streamed fallback" + + requests_seen = mock_anthropic.received_requests() + assert len(requests_seen) == 2, f"Expected 2 backend requests, got {len(requests_seen)}" + # endswith: the gateway may prefix the first user message with the + # injection (INJECT_POLICY_CONTEXT defaults to true); + # the fallback restores the request as it entered the POLICY, so the + # injection prefix is present on both attempts. + assert _message_content(requests_seen[0]).endswith("POLICY-REWRITTEN from the client") + assert "hello" not in _message_content(requests_seen[0]) + assert _message_content(requests_seen[1]).endswith("hello from the client") 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..4ef610520 100644 --- a/tests/luthien_proxy/unit_tests/pipeline/test_anthropic_processor.py +++ b/tests/luthien_proxy/unit_tests/pipeline/test_anthropic_processor.py @@ -2736,3 +2736,353 @@ async def emissions(): webhook.fire_and_forget.assert_called_once() recorder.flush.assert_called() # cleanup completed despite webhook failure + + +def _status_error(status_code: int, message: str = "bad request") -> AnthropicStatusError: + """Build an AnthropicStatusError with the given status code.""" + response = HttpxResponse( + status_code=status_code, + request=HttpxRequest("POST", "https://api.anthropic.com/v1/messages"), + json={"error": {"type": "invalid_request_error", "message": message}}, + ) + return AnthropicStatusError( + message=message, + response=response, + body={"error": {"type": "invalid_request_error", "message": message}}, + ) + + +class TestPassthroughFallback: + """Tests for the opt-in passthrough fallback in _AnthropicPolicyIO. + + Design principle (Trello kRPRjGUx / PR #204): the proxy should never make + things worse than direct API access. When a policy modification causes a + request-shaped upstream 4xx, retry once with the original request — + observably, and never in a way that overrides an intentional policy block. + """ + + ORIGINAL_REQUEST: AnthropicRequest = { + "model": DEFAULT_TEST_MODEL, + "max_tokens": 100, + "messages": [{"role": "user", "content": "hello"}], + } + + RESPONSE: AnthropicResponse = { + "id": "msg_fallback_ok", + "type": "message", + "role": "assistant", + "content": [{"type": "text", "text": "ok"}], + "model": DEFAULT_TEST_MODEL, + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 1, "output_tokens": 1}, + } + + def _make_io( + self, + *, + enabled: bool, + is_streaming: bool = False, + client: MagicMock | None = None, + ) -> tuple[_AnthropicPolicyIO, MagicMock, MagicMock]: + """Build an _AnthropicPolicyIO with mock client + emitter. + + Returns (io, client, emitter). A fresh copy of ORIGINAL_REQUEST is + used as the initial request so tests can mutate/replace it freely. + """ + import copy as _copy + + client = client or MagicMock() + emitter = MagicMock() + io = _AnthropicPolicyIO( + initial_request=_copy.deepcopy(self.ORIGINAL_REQUEST), + anthropic_client=client, + emitter=emitter, + call_id="test-fallback-call", + session_id="sess-fb", + user_id=None, + request_log_recorder=MagicMock(), + is_streaming=is_streaming, + passthrough_fallback_enabled=enabled, + ) + return io, client, emitter + + def _modified_request(self) -> AnthropicRequest: + return { + "model": DEFAULT_TEST_MODEL, + "max_tokens": 100, + "messages": [{"role": "user", "content": "POLICY-MODIFIED"}], + } + + def _fallback_events(self, emitter: MagicMock) -> list: + return [c for c in emitter.record.call_args_list if c.args[1] == "pipeline.passthrough_fallback"] + + # ── non-streaming ──────────────────────────────────────────────────── + + @pytest.mark.asyncio + async def test_modified_request_400_falls_back_to_original(self): + """Modified request 400s -> original request forwarded, failure recorded.""" + io, client, emitter = self._make_io(enabled=True) + client.complete = AsyncMock(side_effect=[_status_error(400), self.RESPONSE]) + + response = await io.complete(self._modified_request()) + + assert response == self.RESPONSE + assert client.complete.call_count == 2 + # Second (fallback) call must carry the ORIGINAL unmodified request. + retry_request = client.complete.call_args_list[1].args[0] + assert retry_request == self.ORIGINAL_REQUEST + # The policy failure is observable, not silently masked. + fallback_events = self._fallback_events(emitter) + assert len(fallback_events) == 1 + payload = fallback_events[0].args[2] + assert payload["status_code"] == 400 + assert "bad request" in payload["error_message"] + + @pytest.mark.asyncio + async def test_disabled_by_default_no_fallback(self): + """With the flag off (default), the 400 propagates and no retry happens.""" + io, client, emitter = self._make_io(enabled=False) + client.complete = AsyncMock(side_effect=_status_error(400)) + + with pytest.raises(AnthropicStatusError): + await io.complete(self._modified_request()) + + assert client.complete.call_count == 1 + assert self._fallback_events(emitter) == [] + + @pytest.mark.asyncio + async def test_unmodified_request_no_fallback(self): + """If the policy didn't change the request, a retry is pointless: direct + API access would fail identically, so the error propagates.""" + io, client, emitter = self._make_io(enabled=True) + client.complete = AsyncMock(side_effect=_status_error(400)) + + with pytest.raises(AnthropicStatusError): + await io.complete() # io._request is untouched == original + + assert client.complete.call_count == 1 + assert self._fallback_events(emitter) == [] + + @pytest.mark.asyncio + @pytest.mark.parametrize("status_code", [401, 403, 429, 500, 529]) + async def test_non_request_shaped_errors_never_fall_back(self, status_code: int): + """Auth, rate-limit, and server errors are not caused by body + modifications; retrying would waste load or mask credential issues.""" + io, client, emitter = self._make_io(enabled=True) + client.complete = AsyncMock(side_effect=_status_error(status_code)) + + with pytest.raises(AnthropicStatusError): + await io.complete(self._modified_request()) + + assert client.complete.call_count == 1 + assert self._fallback_events(emitter) == [] + + @pytest.mark.asyncio + async def test_fallback_retry_failure_propagates(self): + """If the original request ALSO fails, the client sees exactly what + direct API access would return — and the fallback event is still + recorded so the policy failure is not lost.""" + io, client, emitter = self._make_io(enabled=True) + client.complete = AsyncMock( + side_effect=[_status_error(400, "modified bad"), _status_error(400, "original bad")] + ) + + with pytest.raises(AnthropicStatusError, match="original bad"): + await io.complete(self._modified_request()) + + assert client.complete.call_count == 2 + assert len(self._fallback_events(emitter)) == 1 + + @pytest.mark.asyncio + async def test_in_place_policy_mutation_is_detected(self): + """Policies may mutate the request dict in place (same object). The + deepcopy snapshot must still detect the modification and restore the + pristine original.""" + io, client, emitter = self._make_io(enabled=True) + client.complete = AsyncMock(side_effect=[_status_error(400), self.RESPONSE]) + + # Simulate an in-place mutating policy: same dict object, nested edit. + mutated = io.request + mutated["messages"][0]["content"] = "POLICY-MODIFIED-IN-PLACE" + io.set_request(mutated) + + response = await io.complete(mutated) + + assert response == self.RESPONSE + retry_request = client.complete.call_args_list[1].args[0] + assert retry_request["messages"][0]["content"] == "hello" + assert len(self._fallback_events(emitter)) == 1 + + # ── streaming ──────────────────────────────────────────────────────── + + def _stream_events(self) -> list[MessageStreamEvent]: + return [ + RawMessageStartEvent( + type="message_start", + message={ + "id": "msg_stream_fb", + "type": "message", + "role": "assistant", + "content": [], + "model": DEFAULT_TEST_MODEL, + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 1, "output_tokens": 0}, + }, + ), + RawMessageStopEvent(type="message_stop"), + ] + + @pytest.mark.asyncio + async def test_streaming_connect_failure_falls_back(self): + """A 400 at stream connect (zero events yielded) falls back to + streaming the original request.""" + ok_events = self._stream_events() + + async def failing_stream(request, extra_headers=None): + raise _status_error(400) + yield # pragma: no cover — makes this an async generator + + async def ok_stream(request, extra_headers=None): + for event in ok_events: + yield event + + client = MagicMock() + client.stream = MagicMock(side_effect=[failing_stream(None), ok_stream(None)]) + io, client, emitter = self._make_io(enabled=True, is_streaming=True, client=client) + + received = [] + async for event in io.stream(self._modified_request()): + received.append(event) + + assert received == ok_events + assert client.stream.call_count == 2 + retry_request = client.stream.call_args_list[1].args[0] + assert retry_request == self.ORIGINAL_REQUEST + assert len(self._fallback_events(emitter)) == 1 + + @pytest.mark.asyncio + async def test_streaming_mid_stream_error_never_falls_back(self): + """Once events have flowed, re-sending would duplicate content for the + policy/client. Mid-stream errors propagate (no mid-stream recovery).""" + first_event = self._stream_events()[0] + + async def mid_stream_failure(request, extra_headers=None): + yield first_event + raise _status_error(400) + + client = MagicMock() + client.stream = MagicMock(side_effect=[mid_stream_failure(None)]) + io, client, emitter = self._make_io(enabled=True, is_streaming=True, client=client) + + received = [] + with pytest.raises(AnthropicStatusError): + async for event in io.stream(self._modified_request()): + received.append(event) + + assert received == [first_event] + assert client.stream.call_count == 1 + assert self._fallback_events(emitter) == [] + + @pytest.mark.asyncio + async def test_streaming_disabled_no_fallback(self): + """Streaming path also honors the flag being off.""" + + async def failing_stream(request, extra_headers=None): + raise _status_error(400) + yield # pragma: no cover + + client = MagicMock() + client.stream = MagicMock(side_effect=[failing_stream(None)]) + io, client, emitter = self._make_io(enabled=False, is_streaming=True, client=client) + + with pytest.raises(AnthropicStatusError): + async for _ in io.stream(self._modified_request()): + pass + + assert client.stream.call_count == 1 + assert self._fallback_events(emitter) == [] + + # ── intentional blocks are not failures ────────────────────────────── + + @pytest.mark.asyncio + async def test_intentional_response_block_is_untouched(self): + """A policy that blocks by rewriting the response (the ToolCallJudge + pattern) sees no upstream error, so fallback cannot fire: the blocked + content reaches the client and the backend is called exactly once.""" + + class _BlockingPolicy: + async def on_anthropic_request(self, request: AnthropicRequest, context: PolicyContext) -> AnthropicRequest: + return request + + async def on_anthropic_response( + self, response: AnthropicResponse, context: PolicyContext + ) -> AnthropicResponse: + blocked = dict(response) + blocked["content"] = [{"type": "text", "text": "BLOCKED by policy"}] + return blocked # type: ignore[return-value] + + async def on_anthropic_stream_event( + self, event: MessageStreamEvent, context: PolicyContext + ) -> list[MessageStreamEvent]: + return [event] + + async def on_anthropic_stream_complete(self, context: PolicyContext) -> list[AnthropicPolicyEmission]: + return [] + + io, client, emitter = self._make_io(enabled=True) + client.complete = AsyncMock(return_value=dict(self.RESPONSE)) + ctx = make_policy_context() + + emissions = [] + async for emission in _run_policy_hooks(_BlockingPolicy(), io, ctx): + emissions.append(emission) + + assert len(emissions) == 1 + assert emissions[0]["content"][0]["text"] == "BLOCKED by policy" + assert client.complete.call_count == 1 + assert self._fallback_events(emitter) == [] + + @pytest.mark.asyncio + async def test_policy_raised_error_is_not_a_fallback_trigger(self): + """A policy that blocks by raising (fail-secure judge pattern) raises + outside the backend-call site: the error propagates, the backend is + never called, and no fallback fires.""" + + class _RaisingPolicy: + async def on_anthropic_request(self, request: AnthropicRequest, context: PolicyContext) -> AnthropicRequest: + raise RuntimeError("blocked: policy rejected this request") + + async def on_anthropic_response( + self, response: AnthropicResponse, context: PolicyContext + ) -> AnthropicResponse: + return response + + async def on_anthropic_stream_event( + self, event: MessageStreamEvent, context: PolicyContext + ) -> list[MessageStreamEvent]: + return [event] + + async def on_anthropic_stream_complete(self, context: PolicyContext) -> list[AnthropicPolicyEmission]: + return [] + + io, client, emitter = self._make_io(enabled=True) + client.complete = AsyncMock() + ctx = make_policy_context() + + with pytest.raises(RuntimeError, match="blocked"): + async for _ in _run_policy_hooks(_RaisingPolicy(), io, ctx): + pass + + client.complete.assert_not_called() + assert self._fallback_events(emitter) == [] + + def test_snapshot_only_taken_when_enabled(self): + """No deepcopy cost on the default path: no-op stays no-op.""" + io_off, _, _ = self._make_io(enabled=False) + io_on, _, _ = self._make_io(enabled=True) + assert io_off._fallback_original_request is None + assert io_on._fallback_original_request == self.ORIGINAL_REQUEST + # The snapshot is an independent copy, not an alias. + assert io_on._fallback_original_request is not io_on.request