fix(proxy): only drop a batch record on a verdict the guardrail actually reached

A guardrail that reports a technical failure as an HTTPException carrying a block status was
read as a content block, so an unreachable backend under a fail-closed policy quietly shrank the
file instead of failing the upload. Two in-tree integrations do exactly that, and one of them
defaults to fail-closed, so the broken configuration was the default one. Such an exception is
raised `from` the underlying error, which is a deliberate statement that something else caused
it, and no content verdict in the repo is raised that way, so the chain now settles it. Implicit
context is left alone, since a block raised inside an unrelated `except` would read as a failure.

Two annotation errors in the same family: the one GuardrailRaisedException subclass in tree never
opted into blocked_content, so a real block took the whole upload down with it, and straiker's
block helper is reached both from its verdict and from its fail-closed handler, so it claimed a
verdict for an outage. The helper now takes the flag from its caller.

A record could also opt itself out of the chain. Guardrail selection reads a body-level
`guardrails` key ahead of the proxy-injected list, and online that key can only add to the key
and team selection, never replace it, so a batch record naming an empty list skipped every
guardrail that was not default_on and was still reported as scanned. Every injected key is now
stripped before dispatch and restored afterwards.

A guardrail that reroutes a record to another model is honoured on the online path by rewriting
the model, which the scan read as a rewrite and submitted in the same file, sending content to
the provider the reroute existed to avoid. Every record of a batch file goes to one provider, so
the upload is refused instead, naming the line.

The scan spool is closed on the paths that never read it back.

litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py

@@ -54,6 +54,7 @@ class OvalixGuardrailBlockedException(GuardrailRaisedException):
             guardrail_name=guardrail_name,
             message=message,
             should_wrap_with_default_message=should_wrap_with_default_message,
+            blocked_content=True,
         )
 
 

litellm/proxy/guardrails/guardrail_hooks/straiker/straiker.py

@@ -536,13 +536,14 @@ class StraikerGuardrail(CustomGuardrail):
         request_data: dict,
         input_type: Literal["request", "response"],
         message: str,
+        blocked_content: bool = False,
     ) -> NoReturn:
         if input_type == "request":
             raise GuardrailRaisedException(
                 guardrail_name=self.guardrail_name or GUARDRAIL_NAME,
                 message=message,
                 should_wrap_with_default_message=False,
-                blocked_content=True,
+                blocked_content=blocked_content,
             )
         raise ModifyResponseException(
             message=message,
@@ -624,6 +625,7 @@ class StraikerGuardrail(CustomGuardrail):
                 request_data=request_data,
                 input_type=input_type,
                 message=parsed.blocked_reason or DEFAULT_BLOCK_MESSAGE,
+                blocked_content=True,
             )
         if parsed.action == "GUARDRAIL_INTERVENED":
             is_streamed_response: Final = input_type == "response" and _is_streamed_request(request_data)
@@ -632,6 +634,7 @@ class StraikerGuardrail(CustomGuardrail):
                     request_data=request_data,
                     input_type=input_type,
                     message=parsed.blocked_reason or DEFAULT_BLOCK_MESSAGE,
+                    blocked_content=True,
                 )
             return self._intervened_inputs(inputs, parsed)
         return inputs

litellm/proxy/openai_files_endpoints/batch_guardrails.py

@@ -47,10 +47,15 @@ _SUMMARY_LIMIT: Final = 50
 
 _SCAN_METADATA_KEY: Final = "litellm_metadata"
 
-# `metadata` is dropped rather than diffed: guardrail dispatch writes its bookkeeping into it
-# whenever the payload has one, and a record's own metadata is not scanned content on the
-# online path either.
-_INJECTED_KEYS: Final = frozenset({_SCAN_METADATA_KEY, "metadata"})
+# Set by pre_call_hook when a guardrail rerouted the request to a different model.
+_ROUTE_APPLIED_KEY: Final = "sensitive_data_routing_applied"
+
+# Dropped before dispatch and restored afterwards rather than diffed. Guardrail dispatch writes
+# its bookkeeping into `metadata`, and a record's own metadata is not scanned content on the
+# online path either. `guardrails` is dropped because guardrail selection reads it ahead of the
+# proxy-injected list, so leaving it would let a record's own body opt out of the chain its key
+# and team selected; online that key can only add to the list, never replace it.
+_INJECTED_KEYS: Final = frozenset({_SCAN_METADATA_KEY, "metadata", "guardrails"})
 
 # Only what guardrail dispatch reads. The parent OTel span is deliberately left out: parenting one
 # guardrail span per record would put tens of thousands of spans on a single upload's trace.
@@ -97,7 +102,14 @@ class UnscannableRecord:
     url: str | None
 
 
-BatchScanFailure: TypeAlias = UnparseableRecord | UnscannableRecord
+@dataclass(frozen=True, slots=True)
+class UnroutableRecord:
+    line_number: int
+    custom_id: str | None
+    guardrail: str | None
+
+
+BatchScanFailure: TypeAlias = UnparseableRecord | UnscannableRecord | UnroutableRecord
 
 
 @dataclass(frozen=True, slots=True)
@@ -190,6 +202,12 @@ def raise_public(failure: BatchScanFailure) -> NoReturn:
                 "and its body has no messages, prompt or input, so guardrails cannot read it. "
                 "Give the record a chat, completion, embedding, responses or messages body"
             )
+        case UnroutableRecord(line_number=line_number, custom_id=custom_id, guardrail=guardrail):
+            raise _rejected(
+                f"Batch input line {line_number}{_describe(custom_id)} was routed to a different model by "
+                f"{guardrail or 'a guardrail'}, and every record of a batch file goes to one provider, so "
+                "the file cannot be submitted. Send that record outside the batch"
+            )
         case _:
             assert_never(failure)
 
@@ -211,9 +229,16 @@ def _is_content_block(exc: BaseException) -> bool:
     the guardrail to fail closed, so treating it as a block would turn "refuse this request" into
     "drop this record and submit the rest", which is the silent loss of enforcement this whole
     path exists to prevent. A guardrail that does not say it blocked content aborts the upload.
+
+    Guardrails that report a technical failure as an ``HTTPException`` carrying a block status
+    are caught by ``__cause__``: raising ``from`` the underlying error is a deliberate statement
+    that something else caused this, which a verdict on content never is. Implicit context is
+    left alone, since a block raised inside an unrelated ``except`` would read as a failure.
     """
     if isinstance(exc, GuardrailRaisedException):
         return exc.blocked_content
+    if exc.__cause__ is not None:
+        return False
     return is_guardrail_intervention(exc)
 
 
@@ -334,7 +359,8 @@ async def _scan_record(
 
     scan_input: Final[dict[str, object]] = copy.deepcopy(body)  # mutable-ok: pre_call_hook mutates the dict it is given
     own_injected: Final = MappingProxyType({key: body[key] for key in _INJECTED_KEYS if key in body})
-    scan_input.pop("metadata", None)
+    for injected in _INJECTED_KEYS:
+        scan_input.pop(injected, None)
     # Deep, and per record: `headers` and `tags` are nested containers shared with the upload
     # request and with every other record in the window, and a guardrail that writes into one in
     # place would otherwise leak across records and back into the request. The narrowing above
@@ -355,6 +381,14 @@ async def _scan_record(
             return RecordDropped(line_number=record.line_number, custom_id=custom_id, guardrail=_naming_guardrail(exc))
         raise
 
+    rerouted: Final = scanned.get("metadata")
+    if isinstance(rerouted, dict) and rerouted.get(_ROUTE_APPLIED_KEY):
+        return UnroutableRecord(
+            line_number=record.line_number,
+            custom_id=custom_id,
+            guardrail=rerouted.get("sensitive_data_routing_guardrail"),
+        )
+
     compared: Final = (frozenset(body) | frozenset(scanned)) - _INJECTED_KEYS
     if _fingerprint(scanned, compared) == _fingerprint(body, compared):
         return None
@@ -449,14 +483,20 @@ async def scan_batch_input_file(
                     break
         if not problems:
             await drain()
+    except BaseException:
+        redactions.close()
+        raise
     finally:
         file_source.seek(0)
 
     if problems:
+        redactions.close()
         worst: Final = _worst(tuple(problems))
         if isinstance(worst, BaseException):
             raise worst
         return worst
+    if not changes:
+        redactions.close()
     return BatchScanResult(
         changes=tuple(sorted(changes, key=lambda change: change.line_number)),
         scanned_records=sum(scanned),

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_straiker.py

@@ -1067,3 +1067,29 @@ async def test_anthropic_non_streaming_response_reports_usage():
     payload = _posted_payload(g)
     assert payload["usage"] == {"input_tokens": 10, "output_tokens": 5}
     assert payload["response"]["finish_reason"] == "end_turn"
+
+
+def test_fail_closed_backend_failure_is_not_reported_as_a_content_verdict():
+    """A drop-one-record consumer must be able to tell a verdict from an outage; _fail is not a verdict."""
+    from litellm.exceptions import GuardrailRaisedException
+
+    guardrail = _make_guardrail()
+
+    with pytest.raises(GuardrailRaisedException) as unreachable:
+        guardrail._fail(
+            inputs={},
+            request_data={"model": "m"},
+            input_type="request",
+            error="connection refused",
+            is_unreachable=True,
+        )
+    assert unreachable.value.blocked_content is False
+
+    with pytest.raises(GuardrailRaisedException) as verdict:
+        guardrail._block(
+            request_data={"model": "m"},
+            input_type="request",
+            message="blocked",
+            blocked_content=True,
+        )
+    assert verdict.value.blocked_content is True

tests/test_litellm/proxy/openai_files_endpoint/test_batch_guardrails.py

@@ -672,6 +672,143 @@ async def test_an_unreachable_guardrail_aborts_instead_of_quietly_dropping_the_r
         await _scan_full(_jsonl(_record("a"), _record("b", content="tripwire")), FakeProxyLogging(_hook))
 
 
+@pytest.mark.asyncio
+async def test_a_guardrail_subclass_that_blocks_content_drops_only_that_record():
+    """A subclass has to opt in too, or a real block takes the whole upload down with it."""
+    from litellm.proxy.guardrails.guardrail_hooks.ovalix.ovalix import OvalixGuardrailBlockedException
+
+    def _hook(data):
+        if "tripwire" in data["messages"][0]["content"]:
+            raise OvalixGuardrailBlockedException(guardrail_name="ovalix", message="blocked")
+
+    result = await _scan_full(_jsonl(_record("a"), _record("b", content="tripwire")), FakeProxyLogging(_hook))
+
+    assert result.changes == (RecordDropped(line_number=2, custom_id="b", guardrail="ovalix"),)
+    assert result.submitted_records == 1
+
+
+@pytest.mark.asyncio
+async def test_a_record_a_guardrail_rerouted_aborts_rather_than_shipping_to_the_original_provider():
+    """pre_call_hook honours a reroute by rewriting `model`; a batch file cannot follow it."""
+    from litellm.proxy.openai_files_endpoints.batch_guardrails import UnroutableRecord
+
+    def _hook(data):
+        if "tripwire" in data["messages"][0]["content"]:
+            data["model"] = "on-prem-model"
+            data["metadata"] = {
+                "sensitive_data_routing_applied": True,
+                "sensitive_data_routing_guardrail": "router-guard",
+            }
+
+    failure = await _scan(_jsonl(_record("a"), _record("b", content="tripwire")), FakeProxyLogging(_hook))
+
+    assert failure == UnroutableRecord(line_number=2, custom_id="b", guardrail="router-guard")
+    with pytest.raises(HTTPException) as caught:
+        raise_public(failure)
+    assert "routed to a different model" in str(caught.value.detail)
+
+
+@pytest.mark.asyncio
+async def test_the_scan_spool_is_closed_when_nothing_will_read_it():
+    """The spool is opened for every scan, so a clean file must not leave a temp handle behind."""
+    result = await _scan_full(_jsonl(_record("a")), FakeProxyLogging())
+
+    assert result.changes == ()
+    assert result.redactions.closed
+
+
+@pytest.mark.asyncio
+async def test_the_scan_spool_is_closed_when_the_upload_is_refused():
+    def _hook(data):
+        if "tripwire" in data["messages"][0]["content"]:
+            raise RuntimeError("infrastructure is down")
+
+    source = _jsonl(_record("a"), _record("b", content="tripwire"))
+    spools = []
+    import litellm.proxy.openai_files_endpoints.batch_guardrails as bg
+
+    real = bg.tempfile.SpooledTemporaryFile
+
+    def _tracking(*args, **kwargs):
+        handle = real(*args, **kwargs)
+        spools.append(handle)
+        return handle
+
+    bg.tempfile.SpooledTemporaryFile = _tracking
+    try:
+        with pytest.raises(RuntimeError):
+            await _scan_full(source, FakeProxyLogging(_hook))
+    finally:
+        bg.tempfile.SpooledTemporaryFile = real
+
+    assert spools and all(handle.closed for handle in spools)
+
+
+@pytest.mark.asyncio
+async def test_the_scan_spool_is_closed_when_a_record_escapes_the_iterator():
+    """A raise from inside the read loop bypasses the per-record outcome path entirely."""
+    import litellm.proxy.openai_files_endpoints.batch_guardrails as bg
+
+    spools = []
+    real = bg.tempfile.SpooledTemporaryFile
+
+    def _tracking(*args, **kwargs):
+        handle = real(*args, **kwargs)
+        spools.append(handle)
+        return handle
+
+    bg.tempfile.SpooledTemporaryFile = _tracking
+    try:
+        with pytest.raises(json.JSONDecodeError):
+            await _scan_full(io.BytesIO(b"{not json at all}\n"), FakeProxyLogging())
+    finally:
+        bg.tempfile.SpooledTemporaryFile = real
+
+    assert spools and all(handle.closed for handle in spools)
+
+
+@pytest.mark.asyncio
+async def test_a_technical_failure_dressed_as_a_block_status_still_aborts():
+    """xecguard and purview report an unreachable backend as HTTPException(400) under fail-closed."""
+
+    def _hook(data):
+        if "tripwire" in data["messages"][0]["content"]:
+            try:
+                raise ConnectionError("backend unreachable")
+            except ConnectionError as exc:
+                raise HTTPException(
+                    status_code=400, detail={"error": "XecGuard API unreachable (block_on_error=True)"}
+                ) from exc
+
+    with pytest.raises(HTTPException):
+        await _scan_full(_jsonl(_record("a"), _record("b", content="tripwire")), FakeProxyLogging(_hook))
+
+
+@pytest.mark.asyncio
+async def test_a_record_body_cannot_opt_itself_out_of_the_guardrail_chain():
+    """Guardrail selection reads a body-level `guardrails` key first; online it can only add."""
+    seen = []
+
+    await _scan_full(
+        _jsonl({**_record("a"), "body": {**_record("a")["body"], "guardrails": []}}),
+        FakeProxyLogging(lambda d: seen.append(sorted(d))),
+        metadata={"guardrails": ["team-guard"]},
+    )
+
+    assert seen and "guardrails" not in seen[0]
+
+
+@pytest.mark.asyncio
+async def test_a_redacted_record_keeps_its_own_guardrails_key():
+    """Stripping it for the scan must not rewrite what the caller asked the provider to run."""
+    record = _record("m", content="my secret is here")
+    record["body"]["guardrails"] = ["extra-guard"]
+
+    body = await _rewritten_body(record, _redact_containing("secret"))
+
+    assert body["guardrails"] == ["extra-guard"]
+
+
 @pytest.mark.asyncio
 async def test_a_400_that_is_not_a_guardrail_decision_still_aborts():
     """A guardrail's own HTTP client can raise a 400 because OUR payload was rejected, not the content."""