fix(sentry): drop expected upstream provider errors

changelog.d/sentry-expected-upstream-errors.md

@@ -0,0 +1,5 @@
+---
+category: Fixes
+---
+
+**Expected upstream provider errors no longer burn Sentry quota**: the Sentry Anthropic integration captures provider errors unhandled at the SDK call site, before the pipeline converts them into a `BackendAPIError` response. `_sentry_before_send` now drops throttling/availability errors (408, 429, 500, 502, 503, 504, 529) unconditionally. 400/404 — the client sent content or a model name Anthropic legitimately rejected — are dropped only when the request that reached Anthropic is proven to be exactly what the client sent (no policy hook rewrote or mutated it, no `UPSTREAM_HEADERS` template injected a header, no policy-context note was added). 401 requires that **plus** proof the forwarded credential was the client's own rather than the operator's shared `ANTHROPIC_API_KEY`: client-key auth mode always forwards the server's credential, so an unmodified body alone proves nothing about who caused a 401 in that mode, and an invalid operator credential must still report. Both provenance facts are tagged on the Sentry scope at the actual upstream call boundary (`_AnthropicPolicyIO`) as separate tags — `PASSTHROUGH_TAG` (body/headers) and `CREDENTIAL_PASSTHROUGH_TAG` (credential origin) — and checked per status code in `_sentry_before_send`, so a 400/401/404 caused by a proxy bug, a policy, or an invalid operator credential still reports instead of being mistaken for the client's or provider's problem. A status code outside this set (e.g. 403) still reports, so a new upstream failure mode stays visible.

src/luthien_proxy/observability/sentry.py

@@ -2,16 +2,18 @@
 
 Layer 1 (EventScrubber): strips values by key name (api_key, token, etc.)
 Layer 2 (before_send hook): summarizes LLM content variables with type+length,
-strips cookies/server_name, redacts non-safe headers.
+strips cookies/server_name, redacts non-safe headers, and drops expected
+upstream provider errors.
 """
 
 from __future__ import annotations
 
 import logging
 from itertools import islice
-from typing import Any
+from typing import Any, Mapping
 
 import sentry_sdk
+from anthropic import APIStatusError
 from sentry_sdk.integrations.logging import ignore_logger
 from sentry_sdk.scrubber import DEFAULT_DENYLIST, EventScrubber
 from sentry_sdk.types import Event, Hint
@@ -44,6 +46,101 @@
     "raw_http_request",
 }
 
+# Upstream statuses that mean the request/response is the client's or the
+# provider's problem, not a proxy defect. The pipeline already converts every
+# one of these into a BackendAPIError response for the client (see
+# _handle_anthropic_error / _build_error_event, which log at warning and
+# handle every AnthropicStatusError the same way regardless of status code)
+# and, for the throttling/availability codes, the caller retries. They arrive
+# here anyway because the Sentry Anthropic integration captures at the SDK
+# call site with handled=false, before our handler ever sees them.
+
+# 408/429/500/502/503/504/529: provider throttling or brief unavailability.
+# Structurally impossible for the proxy to have provoked — dropped
+# unconditionally (56 unhandled 429 events in three days before this filter
+# existed).
+_PROVIDER_SIDE_STATUS_CODES = frozenset({408, 429, 500, 502, 503, 504, 529})
+
+# 400/404: the client sent content Anthropic legitimately rejected —
+# malformed message content or an unknown model name (LUTHIEN-6: 1,080
+# events for one recurring 400; LUTHIEN-2: unknown-model 404s). This is a
+# property of the request body alone, independent of which credential
+# reached Anthropic. But the proxy is not always a transparent passthrough:
+# policy hooks can mutate or replace the request before it reaches
+# Anthropic, and operator-configured UPSTREAM_HEADERS or policy-context
+# injection can alter it too. Only drop these when the request carries
+# provenance (the PASSTHROUGH_TAG scope tag, set at the actual upstream
+# call boundary in _AnthropicPolicyIO) proving nothing touched it after the
+# client sent it.
+_CONTENT_DEPENDENT_STATUS_CODES = frozenset({400, 404})
+
+# 401: an invalid bearer token passed through client-credential mode
+# (LUTHIEN-D). Unlike 400/404, this is NOT solely a body/header property: in
+# client-key auth mode the *credential* forwarded upstream is the operator's
+# own ANTHROPIC_API_KEY rather than anything the client sent, so an
+# unmodified body proves nothing about whose credential caused the 401 in
+# that mode — an invalid operator credential must still report. Dropping a
+# 401 requires BOTH the PASSTHROUGH_TAG (request untouched) AND the
+# CREDENTIAL_PASSTHROUGH_TAG (credential is the client's own, not the
+# operator's shared key).
+_CREDENTIAL_DEPENDENT_STATUS_CODES = frozenset({401})
+
+_CLIENT_OR_PASSTHROUGH_STATUS_CODES = _CONTENT_DEPENDENT_STATUS_CODES | _CREDENTIAL_DEPENDENT_STATUS_CODES
+
+_EXPECTED_UPSTREAM_STATUS_CODES = _PROVIDER_SIDE_STATUS_CODES | _CLIENT_OR_PASSTHROUGH_STATUS_CODES
+
+# Scope tag set by the Anthropic pipeline (see _AnthropicPolicyIO in
+# anthropic_processor.py) immediately before the upstream call, true only
+# when the request body and headers going to Anthropic are exactly what the
+# client sent — no policy hook, header injection, or context injection
+# touched them. Says nothing about which credential was forwarded; see
+# CREDENTIAL_PASSTHROUGH_TAG for that.
+PASSTHROUGH_TAG = "luthien.request_unmodified_passthrough"
+
+# Scope tag set alongside PASSTHROUGH_TAG, true only when the credential
+# forwarded to Anthropic is the client's own (passthrough / BOTH / explicit
+# x-anthropic-api-key auth) rather than the operator's shared
+# ANTHROPIC_API_KEY substituted in client-key auth mode. Only 401 needs
+# this — a bad body/model name (400/404) is credential-independent.
+CREDENTIAL_PASSTHROUGH_TAG = "luthien.credential_client_supplied"
+
+# The decision table itself: which scope tags (all must be True) are required
+# to drop each expected-upstream status code. Empty for the provider-side
+# codes (unconditional), PASSTHROUGH_TAG alone for the content-dependent
+# codes, both tags for the credential-dependent code. Built from the sets
+# above so it cannot drift from the per-category documentation there, and a
+# status absent from every set above is absent here too, so it always
+# reports — see _is_expected_upstream_error.
+_REQUIRED_TAGS_BY_STATUS: dict[int, tuple[str, ...]] = {
+    **dict.fromkeys(_PROVIDER_SIDE_STATUS_CODES, ()),
+    **dict.fromkeys(_CONTENT_DEPENDENT_STATUS_CODES, (PASSTHROUGH_TAG,)),
+    **dict.fromkeys(_CREDENTIAL_DEPENDENT_STATUS_CODES, (PASSTHROUGH_TAG, CREDENTIAL_PASSTHROUGH_TAG)),
+}
+
+
+def tag_request_provenance(unmodified: bool) -> None:
+    """Record on the current Sentry scope whether the outgoing request is untouched.
+
+    Called at the upstream call boundary so `_sentry_before_send` can tell a
+    genuine client/provider 400/404 from one the proxy or a policy caused.
+    Safe to call even when Sentry is disabled or uninitialized —
+    `sentry_sdk.set_tag` is a no-op against the default scope in that case.
+    """
+    sentry_sdk.set_tag(PASSTHROUGH_TAG, unmodified)
+
+
+def tag_credential_provenance(client_supplied: bool) -> None:
+    """Record on the current Sentry scope whether the forwarded credential is the client's own.
+
+    Called at the upstream call boundary alongside `tag_request_provenance`
+    so `_sentry_before_send` can tell a genuine client credential failure
+    (401) from an invalid operator credential in client-key auth mode. Safe
+    to call even when Sentry is disabled or uninitialized —
+    `sentry_sdk.set_tag` is a no-op against the default scope in that case.
+    """
+    sentry_sdk.set_tag(CREDENTIAL_PASSTHROUGH_TAG, client_supplied)
+
+
 _SAFE_REQUEST_KEYS = {"model", "stream", "max_tokens", "temperature", "top_p", "top_k"}
 _SAFE_HEADERS = {"content-type", "accept", "user-agent", "x-request-id"}
 
@@ -80,6 +177,23 @@ def _summarize(value: Any) -> Any:
     return f"<{type(value).__name__}>"
 
 
+def _is_expected_upstream_error(exc: BaseException | None, tags: Mapping[str, object]) -> bool:
+    """True for provider errors that are the client's or provider's fault, not ours.
+
+    Matches on the SDK exception's own status_code rather than its class so a
+    provider SDK renaming or adding a status subclass cannot silently start
+    reporting again. Looks up the required scope tags for that status in
+    _REQUIRED_TAGS_BY_STATUS and drops only when every one of them is True;
+    a status with no entry there always reports.
+    """
+    if not isinstance(exc, APIStatusError):
+        return False
+    required_tags = _REQUIRED_TAGS_BY_STATUS.get(exc.status_code)
+    if required_tags is None:
+        return False
+    return all(tags.get(tag) is True for tag in required_tags)
+
+
 def _sentry_before_send(event: Event, hint: Hint) -> Event | None:
     """Selectively redact sensitive data while preserving debugging context.
 
@@ -91,8 +205,11 @@ def _sentry_before_send(event: Event, hint: Hint) -> Event | None:
     drop the event entirely, or the (mutated) event to send it.
     """
     exc_info = hint.get("exc_info")
-    if isinstance(exc_info, tuple) and exc_info[0] in {KeyboardInterrupt, SystemExit}:
-        return None
+    if isinstance(exc_info, tuple):
+        if exc_info[0] in {KeyboardInterrupt, SystemExit}:
+            return None
+        if _is_expected_upstream_error(exc_info[1], event.get("tags") or {}):
+            return None
 
     event.pop("server_name", None)
 

src/luthien_proxy/pipeline/anthropic_processor.py

@@ -50,6 +50,7 @@
     build_usage,
 )
 from luthien_proxy.observability.emitter import EventEmitterProtocol
+from luthien_proxy.observability.sentry import tag_credential_provenance, tag_request_provenance
 from luthien_proxy.pipeline.client_format import ClientFormat
 from luthien_proxy.pipeline.policy_context_injection import inject_policy_awareness_anthropic
 from luthien_proxy.pipeline.session import (
@@ -110,10 +111,19 @@ def __init__(
         user_id: str | None,
         request_log_recorder: RequestLogRecorder,
         is_streaming: bool,
+        client_request_unmodified: bool,
+        credential_passthrough: bool,
         extra_headers: dict[str, str] | None = None,
     ) -> None:
         self._request = initial_request
-        self._initial_request = initial_request
+        # Deep-copied: policies are allowed to mutate the request dict
+        # in-place (see the identical rationale on _first_backend_response
+        # below) rather than replacing it via set_request(). A live
+        # reference here would silently "see" that mutation too, corrupting
+        # both this snapshot and the provenance check in
+        # _tag_request_provenance, which compares the request actually sent
+        # upstream against this value to detect policy-side modification.
+        self._initial_request = copy.deepcopy(initial_request)
         self._anthropic_client = anthropic_client
         self._emitter = emitter
         self._call_id = call_id
@@ -122,6 +132,19 @@ def __init__(
         self._request_log_recorder = request_log_recorder
         self._is_streaming = is_streaming
         self._extra_headers = extra_headers
+        # Whether the pipeline had already changed the request/headers before
+        # policy hooks ever ran (context injection, UPSTREAM_HEADERS) — see
+        # process_anthropic_request. Combined with the initial_request
+        # comparison in _tag_provenance to decide passthrough provenance for
+        # Sentry (see observability/sentry.py:PASSTHROUGH_TAG).
+        self._client_request_unmodified = client_request_unmodified
+        # Whether the credential forwarded upstream is the client's own
+        # (passthrough / x-anthropic-api-key auth) rather than the
+        # operator's shared ANTHROPIC_API_KEY substituted in client-key auth
+        # mode — see process_anthropic_request. A 401 caused by an invalid
+        # *operator* credential must never be tagged CREDENTIAL_PASSTHROUGH_TAG
+        # just because the request body was untouched.
+        self._credential_passthrough = credential_passthrough
         self._request_recorded = False
         self._first_backend_response: AnthropicResponse | None = None
         # Raw backend events are only buffered when needed for non-streaming
@@ -182,10 +205,27 @@ def _record_backend_request(self, request: AnthropicRequest) -> None:
             endpoint="/v1/messages",
         )
 
+    def _tag_request_provenance(self, final_request: AnthropicRequest) -> None:
+        """Tag the Sentry scope with the request and credential provenance.
+
+        `self._client_request_unmodified` covers pipeline-level changes made
+        before any policy hook ran (context injection, UPSTREAM_HEADERS);
+        comparing `final_request` against the pre-hook snapshot
+        (`self._initial_request`) covers what a policy hook did to it. Both
+        must hold for PASSTHROUGH_TAG (body/headers untouched). Credential
+        provenance is tagged separately via CREDENTIAL_PASSTHROUGH_TAG since
+        a bad body/model name (400/404) is credential-independent — only a
+        401 needs both tags true (see observability/sentry.py).
+        """
+        unmodified = self._client_request_unmodified and final_request == self._initial_request
+        tag_request_provenance(unmodified)
+        tag_credential_provenance(self._credential_passthrough)
+
     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)
+        self._tag_request_provenance(final_request)
 
         with tracer.start_as_current_span("send_upstream") as span:
             span.set_attribute("luthien.phase", "send_upstream")
@@ -200,6 +240,7 @@ def stream(self, request: AnthropicRequest | None = None) -> AsyncIterator[Messa
         """Execute a streaming backend request."""
         final_request = request or self._request
         self._record_backend_request(final_request)
+        self._tag_request_provenance(final_request)
 
         extra_headers = self._extra_headers
 
@@ -424,13 +465,29 @@ async def process_anthropic_request(
 
         # Expand configurable upstream headers (e.g. Helicone session/auth headers).
         # Templates in UPSTREAM_HEADERS env var are expanded with per-request context.
-        forwarded_headers = merge_forwarded_headers(
-            base=forwarded_headers,
-            upstream=expand_upstream_headers(
-                session_id=session_id,
-                request_path=raw_http_request.path,
-            ),
+        upstream_injected_headers = expand_upstream_headers(
+            session_id=session_id,
+            request_path=raw_http_request.path,
         )
+        forwarded_headers = merge_forwarded_headers(base=forwarded_headers, upstream=upstream_injected_headers)
+
+        # Passthrough provenance so far (before policy hooks run): true only
+        # when neither policy-context injection (above) nor an
+        # operator-configured upstream header changed anything the client
+        # didn't itself send. `anthropic-beta` forwarding doesn't count
+        # against this — it relays a header the client already set, verbatim.
+        # _AnthropicPolicyIO combines this with what happens in policy hooks
+        # to decide the final Sentry PASSTHROUGH_TAG (observability/sentry.py).
+        client_request_unmodified = anthropic_request == raw_http_request.body and not upstream_injected_headers
+
+        # Whether the credential Anthropic will see is the client's own, not
+        # the operator's shared ANTHROPIC_API_KEY. resolve_anthropic_client
+        # (gateway_routes.py) passes `user_credential=None` exactly when a
+        # client-key-mode request matched the shared key and the server's own
+        # base_client/credential is forwarded instead — that branch is never
+        # a client passthrough no matter how untouched the body is, since a
+        # 401 there means the *operator's* credential is invalid.
+        credential_passthrough = user_credential is not None
 
         # Create policy cache factory if database is available. The cap is
         # configured once here so every policy's cache honors the same limit;
@@ -466,6 +523,8 @@ async def process_anthropic_request(
             is_streaming=is_streaming,
             root_span=root_span,
             request_log_recorder=request_log_recorder,
+            client_request_unmodified=client_request_unmodified,
+            credential_passthrough=credential_passthrough,
             extra_headers=forwarded_headers,
             usage_collector=usage_collector,
             webhook_sender=webhook_sender,
@@ -620,6 +679,8 @@ async def _execute_anthropic_policy(
     root_span: Span,
     request_log_recorder: RequestLogRecorder,
     request_start_time: float,
+    client_request_unmodified: bool,
+    credential_passthrough: bool,
     extra_headers: dict[str, str] | None = None,
     usage_collector: UsageCollector | None = None,
     webhook_sender: WebhookSender | None = None,
@@ -634,6 +695,8 @@ async def _execute_anthropic_policy(
         user_id=policy_ctx.user_id,
         request_log_recorder=request_log_recorder,
         is_streaming=is_streaming,
+        client_request_unmodified=client_request_unmodified,
+        credential_passthrough=credential_passthrough,
         extra_headers=extra_headers,
     )
     emissions = _run_policy_hooks(execution_policy, io, policy_ctx)