scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-06 08:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

fix(logging): blocked requests no longer report guardrail_status=success in multi-guardrail configs (#39596)

Browse source 
* fix(logging): aggregate guardrail_status by severity across guardrail entries

A pre_call guardrail that passed (e.g. hide-secrets recording a mask)
appends its entry before a later guardrail's block, and the first-wins
reader reported the blocked request as guardrail_status=success in
StandardLoggingPayload.status_fields. Take the most severe status
across all entries instead: guardrail_intervened >
guardrail_failed_to_respond > success > not_run.

* refactor(logging): express guardrail status severity as an immutable order

Replace the precedence dict and rebinding loop with a severity-ordered
tuple and a max() aggregation, per the repo's no-mutation and
mutable-collection lint gates; parametrize the severity test cases.
No behavior change.

* style(logging): apply ruff format to entries binding
This commit is contained in:
yucheng-berri 2026-09-03 17:33:31 -07:00 • committed by GitHub
parent fe770700f4
commit a06d63f99e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 111 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
litellm/litellm_core_utils/litellm_logging.py
View file

@ -10,7 +10,7 @@ import subprocess
import sys
import time
import traceback
from collections.abc import Callable, Mapping, Sequence
from collections.abc import Callable, Iterator, Mapping, Sequence
from datetime import datetime as dt_object
from functools import lru_cache
from types import MappingProxyType, TracebackType
@ -5878,14 +5878,28 @@ def _get_status_fields(
#########################################################
# Map - guardrail_information.guardrail_status to guardrail_status
#########################################################
guardrail_status: GuardrailStatus = "not_run"
if guardrail_information and isinstance(guardrail_information, list):
for information in guardrail_information:
if isinstance(information, dict):
raw_status = information.get("guardrail_status", "not_run")
if raw_status != "not_run":
guardrail_status = GUARDRAIL_STATUS_MAP.get(raw_status, "not_run")
break
# Severity order, least severe first. The status aggregates across ALL
# guardrail entries rather than taking the first non-"not_run" one: a
# pre_call guardrail that passed (e.g. a mask) records its entry before a
# later guardrail's block, and first-wins would report a blocked request
# as "success".
GUARDRAIL_STATUS_SEVERITY: Final[tuple[GuardrailStatus, ...]] = (
"not_run",
"success",
"guardrail_failed_to_respond",
"guardrail_intervened",
)
entries: Final[Sequence[object]] = guardrail_information if isinstance(guardrail_information, list) else ()
raw_statuses: Final[Iterator[object]] = (
entry.get("guardrail_status", "not_run") for entry in entries if isinstance(entry, dict)
)
# A guardrail is free to write any value here, and an unhashable one would
# raise TypeError on the mapping lookup and drop the whole payload.
guardrail_status: Final[GuardrailStatus] = max(
(GUARDRAIL_STATUS_MAP.get(raw_status, "not_run") for raw_status in raw_statuses if isinstance(raw_status, str)),
key=GUARDRAIL_STATUS_SEVERITY.index,
default="not_run",
)
return StandardLoggingPayloadStatusFields(llm_api_status=llm_api_status, guardrail_status=guardrail_status)

88
tests/guardrails_tests/test_tracing_guardrails.py
View file

@ -806,3 +806,91 @@ def test_guardrail_status_fields_computation():
)
assert status_fields_no_guardrail.get("llm_api_status") == "success"
assert status_fields_no_guardrail.get("guardrail_status") == "not_run"
@pytest.mark.parametrize(
"status, guardrail_information, expected_guardrail_status",
[
pytest.param(
"failure",
[
{"guardrail_status": "success"},
{"guardrail_status": "guardrail_intervened"},
],
"guardrail_intervened",
id="pre_call_success_before_blocker",
),
pytest.param(
"failure",
[
{"guardrail_status": "guardrail_intervened"},
{"guardrail_status": "success"},
],
"guardrail_intervened",
id="blocker_before_success",
),
pytest.param(
"failure",
[
{"guardrail_status": "success"},
{"guardrail_status": "guardrail_failed_to_respond"},
],
"guardrail_failed_to_respond",
id="failure_outranks_success",
),
pytest.param(
"failure",
[
{"guardrail_status": "guardrail_failed_to_respond"},
{"guardrail_status": "guardrail_intervened"},
],
"guardrail_intervened",
id="intervention_outranks_failure",
),
pytest.param(
"success",
[
{"guardrail_status": "success"},
{"guardrail_status": "success"},
],
"success",
id="all_success_stays_success",
),
pytest.param(
"failure",
[
{"guardrail_status": "some_new_status"},
{"guardrail_status": "blocked"},
],
"guardrail_intervened",
id="unknown_status_does_not_mask_blocker",
),
pytest.param(
"failure",
[
{"guardrail_status": {"unhashable": True}},
{"guardrail_status": "guardrail_intervened"},
],
"guardrail_intervened",
id="unhashable_status_is_skipped",
),
],
)
def test_guardrail_status_fields_severity_across_entries(
status, guardrail_information, expected_guardrail_status
):
"""
A blocked request must never be reported as a guardrail success.
With multiple guardrails on one request (e.g. a pre_call mask that passes,
then a post_call guardrail that blocks), entries are recorded in execution
order, so the earlier "success" entry must not shadow the later
"guardrail_intervened" entry: the aggregate takes the most severe status,
regardless of entry order.
"""
from litellm.litellm_core_utils.litellm_logging import _get_status_fields
fields = _get_status_fields(
status=status, guardrail_information=guardrail_information, error_str=None
)
assert fields.get("guardrail_status") == expected_guardrail_status