diff --git a/litellm/constants.py b/litellm/constants.py index 1d5c6404bfd..5181848ee4e 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -493,6 +493,7 @@ LITELLM_CONFIG_CACHE_TTL_SECONDS: Final = get_env_int("LITELLM_CONFIG_PARAM_CACH GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS: Final = get_env_int( "GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS", 24 * 60 * 60 ) +LOGGING_ONLY_VERDICT_TIMEOUT_SECONDS: Final = 3.0 BEDROCK_APPLY_GUARDRAIL_CHUNK_BUDGET_CHARS: Final = 25_000 CONTENT_FILTER_STREAMING_HOLDBACK_CHARS: Final = 50 CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS: Final = 512 diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py index 9e2dc902d55..af1073f18bc 100644 --- a/litellm/integrations/custom_guardrail.py +++ b/litellm/integrations/custom_guardrail.py @@ -9,6 +9,7 @@ from types import MappingProxyType from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, Optional, cast, get_args import httpx +from pydantic import TypeAdapter from litellm._internal_context import with_service_target from litellm._logging import verbose_logger @@ -19,6 +20,7 @@ from litellm.litellm_core_utils.core_helpers import ( get_or_create_metadata_bucket, redact_nested_match_and_regex_keys, ) +from litellm.llms import get_guardrail_translation_mapping from litellm.secret_managers.main import str_to_bool from litellm.types.guardrails import ( DEFAULT_GUARDRAIL_STREAM_SCOPE, @@ -26,6 +28,7 @@ from litellm.types.guardrails import ( GuardrailEventHooks, GuardrailStreamScope, LitellmParams, + LoggingOnlyInputTiming, LoggingOnlyScope, Mode, runtime_stream_scope, @@ -38,6 +41,7 @@ from litellm.types.utils import ( GuardrailStatus, GuardrailTracingDetail, LLMResponseTypes, + ModelResponse, StandardLoggingGuardrailInformation, ) @@ -50,6 +54,7 @@ dc: Final = DualCache() from litellm.constants import ( GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS, + LOGGING_ONLY_VERDICT_TIMEOUT_SECONDS, LOGS_GUARDRAIL_INFORMATION_MARKER, PRE_CALL_EXECUTED_GUARDRAILS_KEY, SERVER_STREAMING_CLASSIFICATION_KEY, @@ -207,6 +212,35 @@ def guardrail_request_data_with_streaming( } +_STRING_KEYED: Final = TypeAdapter(dict[str, object]) +_GUARDRAIL_ENTRIES: Final = TypeAdapter(tuple[object, ...]) + + +def guardrail_entries(recorded: object) -> tuple[object, ...]: + if isinstance(recorded, list): + return _GUARDRAIL_ENTRIES.validate_python(recorded) + return (recorded,) if recorded else () + + +def _logged_proxy_server_request(kwargs: Mapping[str, object]) -> object: + litellm_params: Final = kwargs.get("litellm_params") + if not isinstance(litellm_params, dict): + return None + return _STRING_KEYED.validate_python(litellm_params).get("proxy_server_request") + + +def with_guardrail_information( + kwargs: dict[str, object], # mutable-ok: CustomLogger.async_logging_hook contract + entries: Sequence[object], +) -> dict[str, object]: # mutable-ok: CustomLogger.async_logging_hook contract + standard_logging_object: Final = kwargs.get("standard_logging_object") + if not entries or not isinstance(standard_logging_object, dict): + return kwargs + payload: Final = _STRING_KEYED.validate_python(standard_logging_object) + existing: Final = guardrail_entries(payload.get("guardrail_information")) + return {**kwargs, "standard_logging_object": {**payload, "guardrail_information": [*existing, *entries]}} + + def _request_is_streaming(data: object, event_type: GuardrailEventHooks | None = None) -> bool: if event_type in _REALTIME_STREAMING_HOOKS: return True @@ -228,6 +262,8 @@ class CustomGuardrail(CustomLogger): records_own_guardrail_information: ClassVar[bool] = False logging_only_scope: LoggingOnlyScope | None logging_only_continue_on_input_failure: bool + logging_only_input_timing: LoggingOnlyInputTiming + logging_only_verdict_timeout: float stream_scope_default: GuardrailStreamScope = DEFAULT_GUARDRAIL_STREAM_SCOPE stream_scope_by_hook: tuple[tuple[str, GuardrailStreamScope], ...] = () @@ -311,6 +347,8 @@ class CustomGuardrail(CustomLogger): self.apply_stream_scope(stream_scope_arg) self.logging_only_scope = None self.logging_only_continue_on_input_failure = False + self.logging_only_input_timing = "after_call" + self.logging_only_verdict_timeout = LOGGING_ONLY_VERDICT_TIMEOUT_SECONDS if timeout is not None: self.timeout = timeout @@ -875,6 +913,10 @@ class CustomGuardrail(CustomLogger): def accepts_logging_only_mode(cls) -> bool: return cls.uses_apply_guardrail_interface() and not cls.use_native_lifecycle_hooks + def configure_logging_only_input_timing(self, timing: LoggingOnlyInputTiming, verdict_timeout: float) -> None: + self.logging_only_input_timing = timing + self.logging_only_verdict_timeout = verdict_timeout + @classmethod def supports_logging_only_scope(cls) -> bool: return ( @@ -1016,8 +1058,6 @@ class CustomGuardrail(CustomLogger): call_type: str, ) -> tuple[dict, object]: # mutable-ok: CustomLogger.async_logging_hook contract """logging_only: scan copies of the logged request and/or response according to logging_only_scope.""" - from litellm.llms import get_guardrail_translation_mapping - if not self.uses_apply_guardrail_interface(): return kwargs, result if not self._event_hook_is_event_type(GuardrailEventHooks.logging_only): @@ -1040,8 +1080,6 @@ class CustomGuardrail(CustomLogger): response: Final = ( kwargs.get("async_complete_streaming_response") or kwargs.get("complete_streaming_response") or result ) - from litellm.types.utils import ModelResponse - output_translation: Final = ( get_guardrail_translation_mapping(CallTypes.acompletion)() if isinstance(response, ModelResponse) @@ -1051,16 +1089,10 @@ class CustomGuardrail(CustomLogger): await self._scan_logged_call(kwargs, response, translation, output_translation, scratch_metadata) except Exception as e: verbose_logger.warning("Guardrail %s: logging_only scan raised: %s", self.guardrail_name, e) - recorded: Final = scratch_metadata.get("standard_logging_guardrail_information") - standard_logging_object: Final = kwargs.get("standard_logging_object") - if not recorded or not isinstance(standard_logging_object, dict): - return kwargs, result - entries: Final = recorded if isinstance(recorded, list) else [recorded] - existing: Final = standard_logging_object.get("guardrail_information") or [] - return { - **kwargs, - "standard_logging_object": {**standard_logging_object, "guardrail_information": [*existing, *entries]}, - }, result + return with_guardrail_information( + kwargs, # pyright: ignore[reportUnknownArgumentType] # async_logging_hook's kwargs is the untyped CustomLogger contract + guardrail_entries(scratch_metadata.get("standard_logging_guardrail_information")), + ), result def _copy_scratch_request_fields( self, @@ -1101,6 +1133,7 @@ class CustomGuardrail(CustomLogger): "tools": scratch_tools, "litellm_call_id": kwargs.get("litellm_call_id"), "metadata": scratch_metadata, + "proxy_server_request": _logged_proxy_server_request(kwargs), } if self.logging_only_scope != "output" and scratch_fields is not None: if self.logging_only_continue_on_input_failure: diff --git a/litellm/integrations/logging_only_input_scan.py b/litellm/integrations/logging_only_input_scan.py new file mode 100644 index 00000000000..08464d57502 --- /dev/null +++ b/litellm/integrations/logging_only_input_scan.py @@ -0,0 +1,196 @@ +import asyncio +import itertools +import time +from collections.abc import Coroutine, Iterator, Mapping, Sequence +from types import MappingProxyType +from typing import Final, TypeAlias + +from pydantic import TypeAdapter, ValidationError + +from litellm._logging import verbose_logger +from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL, PROXY_REJECTED_BEFORE_ROUTING_KEY +from litellm.integrations.custom_guardrail import CustomGuardrail +from litellm.litellm_core_utils.core_helpers import independent_snapshot +from litellm.llms import get_guardrail_translation_mapping +from litellm.types.guardrails import GuardrailEventHooks +from litellm.types.utils import CallTypes + +InputScans: TypeAlias = Mapping[CustomGuardrail, "asyncio.Task[object]"] + +_GUARDRAIL_INFORMATION_KEY: Final = "standard_logging_guardrail_information" +_METADATA_BUCKETS: Final = ("metadata", "litellm_metadata") +_SCRATCH_REQUEST_KEYS: Final = frozenset( + {"model", "messages", "input", "tools", "litellm_call_id", *_METADATA_BUCKETS, "proxy_server_request"} +) +_REQUEST: Final = TypeAdapter(dict[str, object]) +_VERDICTS: Final = TypeAdapter(tuple[object, ...]) +_RUNNING_SCANS: Final[set["asyncio.Task[object]"]] = set() # mutable-ok: strong refs keep a scan alive until it ends + + +def _as_request(value: object) -> dict[str, object] | None: # mutable-ok: guardrails record their verdict into it + try: + return _REQUEST.validate_python(value) + except ValidationError: + return None + + +def _without_recorded_verdicts(value: object) -> object: + bucket: Final = _as_request(value) + if bucket is None: + return value + return {key: item for key, item in bucket.items() if key != _GUARDRAIL_INFORMATION_KEY} + + +def _scratch_request(data: Mapping[str, object]) -> dict[str, object]: # mutable-ok: the scan records its verdict here + selected: Final = {key: item for key, item in data.items() if key in _SCRATCH_REQUEST_KEYS} + without_verdicts: Final[dict[str, object]] = { # mutable-ok: independent_snapshot takes a dict + key: _without_recorded_verdicts(item) for key, item in selected.items() + } + snapshot: Final = _as_request(independent_snapshot(without_verdicts)) + return snapshot if snapshot is not None else {} + + +def _bucket_verdicts(bucket: object) -> tuple[object, ...]: + entries: Final = (_as_request(bucket) or {}).get(_GUARDRAIL_INFORMATION_KEY) + try: + return _VERDICTS.validate_python(entries) if isinstance(entries, list) else () + except ValidationError: + return () + + +def _recorded_verdicts(scratch_request: Mapping[str, object]) -> tuple[object, ...]: + return tuple(itertools.chain.from_iterable(_bucket_verdicts(scratch_request.get(key)) for key in _METADATA_BUCKETS)) + + +async def _scan_input( + guardrail: CustomGuardrail, + call_type: CallTypes, + scratch_request: dict[str, object], # mutable-ok: the guardrail records its verdict into the scratch request +) -> tuple[object, ...]: + try: + await get_guardrail_translation_mapping(call_type)().process_input_messages( + data=scratch_request, guardrail_to_apply=guardrail + ) + except Exception as e: # noqa: BLE001 # an observe-only scan never affects the request; its verdict is already recorded + verbose_logger.warning("Guardrail %s: logging_only on_arrival scan raised: %s", guardrail.guardrail_name, e) + return _recorded_verdicts(scratch_request) + + +def _translatable_call_type(call_type: str) -> CallTypes | None: + try: + parsed: Final = CallTypes(call_type) + get_guardrail_translation_mapping(parsed) + except ValueError: + return None + return parsed + + +def _on_arrival_guardrails(callbacks: Sequence[object], request: Mapping[str, object]) -> Iterator[CustomGuardrail]: + for callback in callbacks: + if ( + isinstance(callback, CustomGuardrail) + and callback.logging_only_input_timing == "on_arrival" + and callback.should_run_guardrail(data=request, event_type=GuardrailEventHooks.logging_only) is True + ): + yield callback + + +def logging_only_decision_data( + guardrail: CustomGuardrail, model_call_details: Mapping[str, object] +) -> Mapping[str, object]: + """What a log-time should_run_guardrail(logging_only) check sees. A guardrail that scans through the generic hook + gets the call's metadata buckets at the top level, so key and team opt-outs apply as they do at pre_call (Logging + keeps them under litellm_params). A guardrail with its own logging hook, such as one that masks logs, keeps the + call as is, so an opted-out key never turns its log redaction off.""" + if not guardrail.supports_logging_only_scope(): + return model_call_details + litellm_params: Final = _as_request(model_call_details.get("litellm_params")) or {} + lifted: Final = {bucket: litellm_params[bucket] for bucket in _METADATA_BUCKETS if litellm_params.get(bucket)} + return {**model_call_details, **lifted} + + +def never_reached_a_model(model_call_details: Mapping[str, object]) -> bool: + """Whether the failed call was rejected before any model was called, by a proxy gate or the router.""" + litellm_params: Final = _as_request(model_call_details.get("litellm_params")) or {} + return bool( + model_call_details.get(LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL) + or litellm_params.get(PROXY_REJECTED_BEFORE_ROUTING_KEY) + ) + + +def _log_scan_failure(task: "asyncio.Task[object]") -> None: + if task.cancelled() or task.exception() is None: + return + verbose_logger.warning("%s raised: %s", task.get_name(), task.exception()) + + +def spawn_scan(coroutine: Coroutine[object, object, object], name: str) -> "asyncio.Task[object]": + task: Final = asyncio.get_running_loop().create_task(coroutine, name=name) + _RUNNING_SCANS.add(task) + task.add_done_callback(_RUNNING_SCANS.discard) + task.add_done_callback(_log_scan_failure) + return task + + +def start_input_scans(callbacks: Sequence[object], data: object, call_type: str) -> InputScans: + """Send the request to every on_arrival logging_only guardrail in the background, without waiting.""" + request: Final = _as_request(data) + guardrails: Final = tuple(_on_arrival_guardrails(callbacks, request)) if request is not None else () + translatable: Final = _translatable_call_type(call_type) if guardrails else None + if request is None or translatable is None: + return MappingProxyType({}) + return MappingProxyType( + { + guardrail: spawn_scan( + _scan_input(guardrail, translatable, _scratch_request(request)), + f"Guardrail {guardrail.guardrail_name} logging_only on_arrival scan", + ) + for guardrail in guardrails + } + ) + + +def _not_ready_verdict(guardrail: CustomGuardrail, reason: str) -> tuple[object, ...]: + scratch_request: Final[dict[str, object]] = {"metadata": {}} # mutable-ok: the guardrail records its entry here + recorded_at: Final = time.time() + guardrail.add_standard_logging_guardrail_information_to_request_data( + guardrail_json_response=reason, + request_data=scratch_request, + guardrail_status="not_run", + start_time=recorded_at, + end_time=recorded_at, + duration=0.0, + event_type=GuardrailEventHooks.logging_only, + ) + return _recorded_verdicts(scratch_request) + + +def _timed_out(guardrail: CustomGuardrail) -> str: + return f"no verdict within {guardrail.logging_only_verdict_timeout:g}s" + + +def _finished_verdicts(guardrail: CustomGuardrail, scan: "asyncio.Task[object]") -> tuple[object, ...]: + if scan.cancelled(): + return _not_ready_verdict(guardrail, _timed_out(guardrail)) + if scan.exception() is not None: + return () + verdicts: Final = scan.result() + return _VERDICTS.validate_python(verdicts) if isinstance(verdicts, tuple) else () + + +async def collect_input_scan(guardrail: CustomGuardrail, scan: "asyncio.Task[object]") -> tuple[object, ...]: + """The scan's verdicts, waiting at most the guardrail's verdict timeout and cancelling the scan after it.""" + if not scan.done(): + await asyncio.wait((scan,), timeout=guardrail.logging_only_verdict_timeout) + if not scan.done(): + scan.cancel() + return _not_ready_verdict(guardrail, _timed_out(guardrail)) + return _finished_verdicts(guardrail, scan) + + +def peek_input_scan(guardrail: CustomGuardrail, scan: "asyncio.Task[object]") -> tuple[object, ...]: + """The scan's verdicts if it already finished, without waiting or cancelling, so a failed attempt never delays + its error or a retry, and the scan stays alive for the attempt that succeeds.""" + if not scan.done(): + return _not_ready_verdict(guardrail, "no verdict yet when the call failed") + return _finished_verdicts(guardrail, scan) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 3b8157b2249..a3dd3cc31d8 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -63,10 +63,18 @@ from litellm.exceptions import ( from litellm.integrations.agentops import AgentOps from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook from litellm.integrations.arize.arize import ArizeLogger -from litellm.integrations.custom_guardrail import CustomGuardrail +from litellm.integrations.custom_guardrail import CustomGuardrail, with_guardrail_information from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.deepeval.deepeval import DeepEvalLogger from litellm.integrations.langtrace import langtrace_trace_endpoint +from litellm.integrations.logging_only_input_scan import ( + InputScans, + collect_input_scan, + logging_only_decision_data, + never_reached_a_model, + peek_input_scan, + spawn_scan, +) from litellm.integrations.mlflow import MlflowLogger from litellm.integrations.sqs import SQSLogger from litellm.litellm_core_utils.classifier_logging import ( @@ -140,6 +148,7 @@ from litellm.types.agents import LiteLLMSendMessageResponse from litellm.types.containers.main import ContainerObject from litellm.types.decisions import DecisionsResponse, OpenAIDecisionResponse from litellm.types.extraction import ExtractionResponse +from litellm.types.guardrails import GuardrailEventHooks from litellm.types.integrations.s3_v2 import S3PartitionGranularity from litellm.types.interactions import ( InteractionsAPIResponse, @@ -641,6 +650,7 @@ class Logging(LiteLLMLoggingBaseClass): baseline_cache_context: "BaselineCacheContext | None" = None baseline_observation: "CapturedBaselineObservation | None" = None _on_failure_payload_built: "Callable[[], None] | None" = None + logging_only_input_scans: InputScans = MappingProxyType({}) @property def _defer_async_logging(self) -> bool: @@ -2487,6 +2497,59 @@ class Logging(LiteLLMLoggingBaseClass): self.record_partial_usage_for_failure(spend.usage, spend.cost) self.cost_breakdown = spend.cost_breakdown + def track_logging_only_input_scans(self, scans: InputScans) -> None: + if scans: + self.logging_only_input_scans = MappingProxyType({**self.logging_only_input_scans, **scans}) + + async def _run_guardrail_logging_hook( + self, guardrail: CustomGuardrail, result: object + ) -> tuple[dict[str, object], object]: # mutable-ok: model_call_details is the logging kwargs dict + input_scan: Final = self.logging_only_input_scans.get(guardrail) + if input_scan is not None and guardrail.logging_only_input_timing == "on_arrival": + return with_guardrail_information( + self.model_call_details, await collect_input_scan(guardrail, input_scan) + ), result + return await guardrail.async_logging_hook( + kwargs=self.model_call_details, + result=result, + call_type=self.call_type, # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # call_type is untyped on Logging + ) + + def _run_logging_only_guardrails_on_failure(self, callbacks: Sequence[object]) -> None: + """Send a failed LLM call's request to generic logging_only guardrails without delaying the error or a retry. + + Proxy-only rejections (auth, rate limits, budgets, guardrail blocks) never reached a model and are skipped, so an + unauthenticated caller cannot make the proxy send its text to a vendor. An on_arrival verdict is logged only if it + is already back; an after_call scan runs in the background and its verdict is not logged.""" + if never_reached_a_model(self.model_call_details): + return + for callback in dict.fromkeys(callbacks): + if not isinstance(callback, CustomGuardrail) or not callback.supports_logging_only_scope(): + continue + if ( + callback.should_run_guardrail( # pyright: ignore[reportUnknownMemberType] # data is untyped on every guardrail override + data=logging_only_decision_data(callback, self.model_call_details), + event_type=GuardrailEventHooks.logging_only, + ) + is not True + ): + continue + if ( + input_scan := self.logging_only_input_scans.get(callback) + ) is not None and callback.logging_only_input_timing == "on_arrival": + self.model_call_details = with_guardrail_information( + self.model_call_details, peek_input_scan(callback, input_scan) + ) + continue + spawn_scan( + callback.async_logging_hook( # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # untyped kwargs contract + kwargs=dict(self.model_call_details), + result=None, + call_type=self.call_type, # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # call_type is untyped on Logging + ), + f"Guardrail {callback.guardrail_name} logging_only scan of a failed call", + ) + def attempt_scoped_copy(self) -> "Logging": """A copy with its own model_call_details, so logging one attempt of a request through it leaves the state the request's other attempts share untouched.""" @@ -3375,11 +3438,9 @@ class Logging(LiteLLMLoggingBaseClass): hook_callbacks: Final = tuple(dict.fromkeys((*callbacks, *delivery_callbacks))) for callback in hook_callbacks: if isinstance(callback, CustomGuardrail): - from litellm.types.guardrails import GuardrailEventHooks - if ( callback.should_run_guardrail( - data=call_details, + data=logging_only_decision_data(callback, call_details), event_type=GuardrailEventHooks.logging_only, ) is not True @@ -3971,11 +4032,9 @@ class Logging(LiteLLMLoggingBaseClass): for callback in hook_callbacks: try: if isinstance(callback, CustomGuardrail): - from litellm.types.guardrails import GuardrailEventHooks - if ( callback.should_run_guardrail( - data=self.model_call_details, + data=logging_only_decision_data(callback, self.model_call_details), event_type=GuardrailEventHooks.logging_only, ) is not True @@ -3983,11 +4042,7 @@ class Logging(LiteLLMLoggingBaseClass): continue self.shadow_eval_request_snapshot = None - self.model_call_details, result = await callback.async_logging_hook( - kwargs=self.model_call_details, - result=result, - call_type=self.call_type, - ) + self.model_call_details, result = await self._run_guardrail_logging_hook(callback, result) elif isinstance(callback, CustomLogger): result = redact_message_input_output_from_custom_logger( result=result, litellm_logging_obj=self, custom_logger=callback @@ -4042,6 +4097,7 @@ class Logging(LiteLLMLoggingBaseClass): ) if isinstance(callback, CustomLogger): # custom logger class + # Lazy: a module-level import of shadow_eval_logger is circular with this module from litellm.integrations.shadow_eval_logger import ShadowEvalLogger model_call_details: dict = self.model_call_details @@ -4498,12 +4554,13 @@ class Logging(LiteLLMLoggingBaseClass): start_time=start_time, end_time=end_time, ) - if self._on_failure_payload_built is not None: - self._on_failure_payload_built() callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_async_failure_callbacks, global_callbacks=litellm._async_failure_callback, ) + self._run_logging_only_guardrails_on_failure(callbacks) + if self._on_failure_payload_built is not None: + self._on_failure_payload_built() result: Final = None # result sent to all loggers, init this to None incase it's not created diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index b10b5ae98b5..92dee42f690 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -15547,6 +15547,22 @@ "description": "when True, a flagged or raising logging_only request scan is logged and the response is still scanned, so both verdicts land. Only applies to mode logging_only and is ignored when logging_only_scope is 'input' or 'output'.", "title": "Logging Only Continue On Input Failure" }, + "logging_only_input_timing": { + "anyOf": [ + { + "enum": [ + "after_call", + "on_arrival" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "description": "when a logging_only request scan runs: 'after_call' (default) scans a copy of the logged request once the call ends; 'on_arrival' sends the request to the guardrail in the background once it passes the pre_call guardrails, in parallel with the model call, and the call's log records the verdict. 'on_arrival' needs mode logging_only, logging_only_scope 'input' and a guardrail without its own logging hook; otherwise API writes are rejected and a config file value is ignored with an error log. Failed model calls are scanned too, without delaying the error or a retry: an 'on_arrival' verdict is logged only if it is already back, and an 'after_call' scan runs in the background without a logged verdict. Requests the proxy rejects before reaching a model (auth, rate limits, budgets, guardrail blocks) are not scanned.", + "title": "Logging Only Input Timing" + }, "logging_only_scope": { "anyOf": [ { @@ -15563,6 +15579,19 @@ "description": "which direction a logging_only scan observes: 'input' (request) or 'output' (response); unset scans both directions. Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking.", "title": "Logging Only Scope" }, + "logging_only_verdict_timeout": { + "anyOf": [ + { + "minimum": 1.0, + "type": "number" + }, + { + "type": "null" + } + ], + "description": "seconds a successful call's log waits for an 'on_arrival' verdict that is not back yet (default 3, minimum 1). When it runs out, the scan is cancelled and logged as not_run. A failed call's log never waits. Only applies with logging_only_input_timing 'on_arrival'.", + "title": "Logging Only Verdict Timeout" + }, "mask_request_content": { "anyOf": [ { @@ -18081,6 +18110,22 @@ "description": "when True, a flagged or raising logging_only request scan is logged and the response is still scanned, so both verdicts land. Only applies to mode logging_only and is ignored when logging_only_scope is 'input' or 'output'.", "title": "Logging Only Continue On Input Failure" }, + "logging_only_input_timing": { + "anyOf": [ + { + "enum": [ + "after_call", + "on_arrival" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "description": "when a logging_only request scan runs: 'after_call' (default) scans a copy of the logged request once the call ends; 'on_arrival' sends the request to the guardrail in the background once it passes the pre_call guardrails, in parallel with the model call, and the call's log records the verdict. 'on_arrival' needs mode logging_only, logging_only_scope 'input' and a guardrail without its own logging hook; otherwise API writes are rejected and a config file value is ignored with an error log. Failed model calls are scanned too, without delaying the error or a retry: an 'on_arrival' verdict is logged only if it is already back, and an 'after_call' scan runs in the background without a logged verdict. Requests the proxy rejects before reaching a model (auth, rate limits, budgets, guardrail blocks) are not scanned.", + "title": "Logging Only Input Timing" + }, "logging_only_scope": { "anyOf": [ { @@ -18097,6 +18142,19 @@ "description": "which direction a logging_only scan observes: 'input' (request) or 'output' (response); unset scans both directions. Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking.", "title": "Logging Only Scope" }, + "logging_only_verdict_timeout": { + "anyOf": [ + { + "minimum": 1.0, + "type": "number" + }, + { + "type": "null" + } + ], + "description": "seconds a successful call's log waits for an 'on_arrival' verdict that is not back yet (default 3, minimum 1). When it runs out, the scan is cancelled and logged as not_run. A failed call's log never waits. Only applies with logging_only_input_timing 'on_arrival'.", + "title": "Logging Only Verdict Timeout" + }, "mask": { "anyOf": [ { diff --git a/litellm/proxy/guardrails/guardrail_endpoints.py b/litellm/proxy/guardrails/guardrail_endpoints.py index a5822bfb0e3..3a3d40a623e 100644 --- a/litellm/proxy/guardrails/guardrail_endpoints.py +++ b/litellm/proxy/guardrails/guardrail_endpoints.py @@ -91,6 +91,7 @@ router: Final = APIRouter() GUARDRAIL_REGISTRY: Final = GuardrailRegistry() _DECODED_JSON: Final = TypeAdapter(object) _DECODED_YAML: Final = TypeAdapter(object) +_LOGGING_ONLY_INPUT_TIMING_FIELDS: Final = ("logging_only_input_timing", "logging_only_verdict_timeout") def _as_str_object_mapping(mapping: Mapping[str, object]) -> Mapping[str, object]: @@ -1307,6 +1308,13 @@ async def patch_guardrail( and mode_lacks_logging_only else {} ), + **{ + field: None + for field in _LOGGING_ONLY_INPUT_TIMING_FIELDS + if mode_lacks_logging_only + and field not in requested_litellm_params + and getattr(parsed_litellm_params, field) is not None + }, } ) litellm_params: Final = ( @@ -1343,7 +1351,8 @@ async def patch_guardrail( IN_MEMORY_GUARDRAIL_HANDLER.sync_guardrail_from_db( guardrail=guardrail, reject_invalid_logging_only_scope=bool( - {"logging_only_scope", "logging_only_continue_on_input_failure"} & set(requested_litellm_params) + {"logging_only_scope", "logging_only_continue_on_input_failure", *_LOGGING_ONLY_INPUT_TIMING_FIELDS} + & set(requested_litellm_params) ), ) verbose_proxy_logger.info( diff --git a/litellm/proxy/guardrails/guardrail_registry.py b/litellm/proxy/guardrails/guardrail_registry.py index 24b30e65e2b..7b9523b752b 100644 --- a/litellm/proxy/guardrails/guardrail_registry.py +++ b/litellm/proxy/guardrails/guardrail_registry.py @@ -15,7 +15,11 @@ import litellm from litellm import Router from litellm._logging import verbose_proxy_logger from litellm._uuid import uuid -from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH, GUARDRAIL_ROTATION_ATTEMPTS +from litellm.constants import ( + DEFAULT_MAX_RECURSE_DEPTH, + GUARDRAIL_ROTATION_ATTEMPTS, + LOGGING_ONLY_VERDICT_TIMEOUT_SECONDS, +) from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.llms.base_llm.guardrail_translation.utils import ( @@ -53,6 +57,7 @@ from litellm.types.guardrails import ( GuardrailEventHooks, LakeraCategoryThresholds, LitellmParams, + LoggingOnlyScope, SupportedGuardrailIntegrations, with_tolerated_stream_scope, ) @@ -617,6 +622,54 @@ def _logging_only_continue_error(guardrail_name: str, litellm_params: LitellmPar return None +def _logging_only_input_timing_error( + custom_guardrail_callback: CustomGuardrail, + guardrail_name: str, + litellm_params: LitellmParams, + logging_only_scope: LoggingOnlyScope | None, +) -> str | None: + if litellm_params.logging_only_input_timing != "on_arrival": + if litellm_params.logging_only_verdict_timeout is None: + return None + return ( + f"Guardrail {guardrail_name}: logging_only_verdict_timeout is set, but logging_only_input_timing is " + "not on_arrival, so it would never apply. Set logging_only_input_timing: on_arrival or remove " + "logging_only_verdict_timeout." + ) + if GuardrailEventHooks.logging_only.value not in configured_event_hooks(litellm_params.mode): + return ( + f"Guardrail {guardrail_name}: logging_only_input_timing is on_arrival, but mode does not include " + "logging_only, so it would never apply. Add logging_only to mode or remove logging_only_input_timing." + ) + if not custom_guardrail_callback.supports_logging_only_scope(): + return ( + f"Guardrail {guardrail_name}: logging_only_input_timing=on_arrival is not supported by this guardrail, " + "whose logging_only hook scans on its own. Remove logging_only_input_timing." + ) + if logging_only_scope != "input": + return ( + f"Guardrail {guardrail_name}: logging_only_input_timing=on_arrival only applies with " + "logging_only_scope: input. Set logging_only_scope: input or remove logging_only_input_timing." + ) + return None + + +def _logging_only_errors( + custom_guardrail_callback: CustomGuardrail, guardrail_name: str, litellm_params: LitellmParams +) -> tuple[str | None, str | None, str | None]: + scope_error: Final = _logging_only_scope_error(custom_guardrail_callback, guardrail_name, litellm_params) + return ( + scope_error, + _logging_only_continue_error(guardrail_name, litellm_params), + _logging_only_input_timing_error( + custom_guardrail_callback, + guardrail_name, + litellm_params, + None if scope_error is not None else litellm_params.logging_only_scope, + ), + ) + + def _configure_callback_scoping( custom_guardrail_callback: CustomGuardrail, guardrail_name: str, @@ -624,12 +677,11 @@ def _configure_callback_scoping( *, reject_invalid_logging_only_scope: bool = False, ) -> None: - logging_only_scope_error: Final = _logging_only_scope_error( + logging_only_scope_error, logging_only_continue_error, logging_only_input_timing_error = _logging_only_errors( custom_guardrail_callback, guardrail_name, litellm_params ) - logging_only_continue_error: Final = _logging_only_continue_error(guardrail_name, litellm_params) if reject_invalid_logging_only_scope: - for error in (logging_only_scope_error, logging_only_continue_error): + for error in (logging_only_scope_error, logging_only_continue_error, logging_only_input_timing_error): if error is not None: raise ValueError(error) if logging_only_scope_error is not None: @@ -642,6 +694,12 @@ def _configure_callback_scoping( "%s Ignoring logging_only_continue_on_input_failure; the guardrail keeps its configured mode.", logging_only_continue_error.replace("\r", "").replace("\n", ""), ) + if logging_only_input_timing_error is not None: + verbose_proxy_logger.error( + "%s Ignoring logging_only_input_timing and logging_only_verdict_timeout; the request is scanned after " + "the call.", + logging_only_input_timing_error.replace("\r", "").replace("\n", ""), + ) logging_only_scope: Final = None if logging_only_scope_error is not None else litellm_params.logging_only_scope custom_guardrail_callback.logging_only_scope = logging_only_scope custom_guardrail_callback.logging_only_continue_on_input_failure = ( @@ -649,6 +707,9 @@ def _configure_callback_scoping( and logging_only_scope is None and litellm_params.logging_only_continue_on_input_failure is True ) + on_arrival: Final = ( + logging_only_input_timing_error is None and litellm_params.logging_only_input_timing == "on_arrival" + ) for scoping_param in ( "skip_system_message_in_guardrail", "skip_tool_message_in_guardrail", @@ -670,6 +731,12 @@ def _configure_callback_scoping( ) if isinstance(custom_guardrail_callback, CustomGuardrail): # pyright: ignore[reportUnnecessaryIsInstance] # module-path classes may only subclass CustomLogger custom_guardrail_callback.apply_stream_scope(litellm_params.stream_scope) + custom_guardrail_callback.configure_logging_only_input_timing( + "on_arrival" if on_arrival else "after_call", + litellm_params.logging_only_verdict_timeout + if on_arrival and litellm_params.logging_only_verdict_timeout is not None + else LOGGING_ONLY_VERDICT_TIMEOUT_SECONDS, + ) _apply_configured_bool_overrides(custom_guardrail_callback, litellm_params) @@ -684,7 +751,12 @@ def parse_tolerant_litellm_params( try: return params_model(**litellm_params_data) except ValidationError as validation_error: - tolerated_fields: Final = ("logging_only_scope", "logging_only_continue_on_input_failure") + tolerated_fields: Final = ( + "logging_only_scope", + "logging_only_continue_on_input_failure", + "logging_only_input_timing", + "logging_only_verdict_timeout", + ) error_locs: Final = {tuple(error["loc"]) for error in validation_error.errors()} tolerated_locs: Final = frozenset((field,) for field in tolerated_fields) if any(loc not in tolerated_locs for loc in error_locs): @@ -871,10 +943,7 @@ class InMemoryGuardrailHandler: litellm_params: Final = LitellmParams(**params) if isinstance(params, dict) else params guardrail_name: Final = guardrail.get("guardrail_name", "Unknown") for custom_guardrail_callback in self._tracked_callbacks(guardrail_id): - for error in ( - _logging_only_scope_error(custom_guardrail_callback, guardrail_name, litellm_params), - _logging_only_continue_error(guardrail_name, litellm_params), - ): + for error in _logging_only_errors(custom_guardrail_callback, guardrail_name, litellm_params): if error is not None: raise ValueError(error) diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 0ddfc5ce62f..7699ffdf159 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -148,6 +148,7 @@ from litellm.integrations.custom_guardrail import ( ModifyResponseException, ) from litellm.integrations.custom_logger import CustomLogger +from litellm.integrations.logging_only_input_scan import start_input_scans from litellm.integrations.prometheus import PrometheusLogger from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting from litellm.integrations.SlackAlerting.utils import add_langfuse_trace_id_to_alert @@ -259,6 +260,7 @@ from litellm.proxy.policy_engine.pipeline_executor import ( ) from litellm.proxy.policy_engine.policy_registry import get_policy_registry from litellm.proxy.policy_engine.policy_resolver import PolicyResolver +from litellm.proxy.prompts.prompt_registry import parse_prompt_version from litellm.repositories.budget_repository import BudgetRepository from litellm.repositories.config_repository import ConfigRepository from litellm.repositories.table_repositories import ( @@ -2959,8 +2961,6 @@ class ProxyLogging: and prompt_id is not None and (call_type == "completion" or call_type == "acompletion" or call_type == "aresponses") ): - from litellm.proxy.prompts.prompt_registry import parse_prompt_version - await self._process_prompt_template( data=data, litellm_logging_obj=litellm_logging_obj, @@ -3117,6 +3117,15 @@ class ProxyLogging: call_type=call_type, ) + if data is not None and litellm_logging_obj is not None and not (skip_guardrails or guardrails_only): + litellm_logging_obj.track_logging_only_input_scans( + start_input_scans( + caps.resolved_callbacks, + data, # pyright: ignore[reportUnknownArgumentType] # pre_call_hook's data is the untyped request dict + call_type, + ) + ) + if data is not None: self._process_guardrail_metadata(data) diff --git a/litellm/types/guardrails.py b/litellm/types/guardrails.py index e1b891caaaf..09adca66267 100644 --- a/litellm/types/guardrails.py +++ b/litellm/types/guardrails.py @@ -1004,6 +1004,7 @@ def runtime_stream_scope( LoggingOnlyScope = Literal["input", "output"] +LoggingOnlyInputTiming: TypeAlias = Literal["after_call", "on_arrival"] def _is_blank(value: object) -> bool: @@ -1332,6 +1333,31 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up ), ) + logging_only_input_timing: LoggingOnlyInputTiming | None = Field( + default=None, + description=( + "when a logging_only request scan runs: 'after_call' (default) scans a copy of the logged request once " + "the call ends; 'on_arrival' sends the request to the guardrail in the background once it passes the " + "pre_call guardrails, in parallel with the model call, and the call's log records the verdict. " + "'on_arrival' needs mode logging_only, logging_only_scope 'input' and a guardrail without its own " + "logging hook; otherwise API writes are rejected and a config file value is ignored with an error log. " + "Failed model calls are scanned too, without delaying the error or a retry: an 'on_arrival' verdict is " + "logged only if it is already back, and an 'after_call' scan runs in the background without a logged " + "verdict. Requests the proxy rejects before reaching a model (auth, rate limits, budgets, guardrail " + "blocks) are not scanned." + ), + ) + + logging_only_verdict_timeout: float | None = Field( + default=None, + ge=1, + description=( + "seconds a successful call's log waits for an 'on_arrival' verdict that is not back yet (default 3, " + "minimum 1). When it runs out, the scan is cancelled and logged as not_run. A failed call's log never " + "waits. Only applies with logging_only_input_timing 'on_arrival'." + ), + ) + @model_validator(mode="before") # pyright: ignore[reportUntypedFunctionDecorator] # pydantic decorators are untyped @classmethod def blank_values_mean_field_default(cls, data: object) -> object: diff --git a/tests/unit/integrations/test_logging_only_input_scan.py b/tests/unit/integrations/test_logging_only_input_scan.py new file mode 100644 index 00000000000..0859a30b198 --- /dev/null +++ b/tests/unit/integrations/test_logging_only_input_scan.py @@ -0,0 +1,444 @@ +import asyncio +import datetime +from collections.abc import Iterator +from typing import Final + +import pytest +from fastapi import HTTPException + +import litellm +from litellm.caching.dual_cache import DualCache +from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL +from litellm.integrations.custom_guardrail import CustomGuardrail, log_guardrail_information +from litellm.integrations.custom_logger import CustomLogger +from litellm.integrations.logging_only_input_scan import collect_input_scan +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.utils import ProxyLogging +from litellm.router_utils.embedding_batch_size_check import async_log_rejection_before_provider_call +from litellm.types.guardrails import GuardrailEventHooks +from litellm.types.utils import GenericGuardrailAPIInputs +from litellm.utils import Rules, function_setup + +PROMPT: Final = "hello from the user" +TENANT_HEADER: Final = {"x-tenant-id": "tenant-7"} + + +class _ScanRecorder(CustomGuardrail): + def __init__(self, gate: asyncio.Event | None = None, block: bool = False) -> None: + super().__init__(guardrail_name="audit-vendor", event_hook=GuardrailEventHooks.logging_only, default_on=True) + self.gate = gate + self.block = block + self.scans: list[tuple[str, list[str], object]] = [] + self.cancelled = False + + @log_guardrail_information + async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None): + self.scans.append( + ( + input_type, + list(inputs.get("texts") or []), + (request_data.get("proxy_server_request") or {}).get("headers"), + ) + ) + if self.gate is not None: + try: + await self.gate.wait() + except asyncio.CancelledError: + self.cancelled = True + raise + if self.block: + raise ValueError("vendor says no") + return GenericGuardrailAPIInputs(texts=["[MASKED]" for _ in inputs.get("texts") or []]) + + +class _OwnLoggingHookGuardrail(_ScanRecorder): + async def async_logging_hook(self, kwargs, result, call_type): + self.scans.append(("own-hook", [], None)) + return kwargs, result + + +class _PreCallObserver(CustomLogger): + async def async_pre_call_hook(self, user_api_key_dict, cache, data, call_type): + return data + + +class _LogRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.logged = asyncio.Event() + self.events: list[tuple[str | None, list[dict]]] = [] + + @property + def status(self) -> str | None: + return self.events[-1][0] + + @property + def guardrail_information(self) -> list[dict]: + return self.events[-1][1] + + def _record(self, kwargs: dict) -> None: + payload: Final = kwargs.get("standard_logging_object") or {} + self.events.append((payload.get("status"), list(payload.get("guardrail_information") or []))) + self.logged.set() + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + self._record(kwargs) + + async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): + self._record(kwargs) + + +def _on_arrival(guardrail: _ScanRecorder, timeout: float = 5.0) -> _ScanRecorder: + guardrail.logging_only_scope = "input" + guardrail.logging_only_input_timing = "on_arrival" + guardrail.logging_only_verdict_timeout = timeout + return guardrail + + +@pytest.fixture +def registered(monkeypatch: pytest.MonkeyPatch) -> Iterator[list[object]]: + callbacks: Final[list[object]] = [] + for name in ( + "callbacks", + "success_callback", + "failure_callback", + "_async_success_callback", + "_async_failure_callback", + ): + monkeypatch.setattr(litellm, name, [] if name != "callbacks" else callbacks) + ProxyLogging._callback_capabilities_cache.clear() + yield callbacks + ProxyLogging._callback_capabilities_cache.clear() + + +def _proxy_request() -> dict: + logging_obj, data = function_setup( + original_function="acompletion", + rules_obj=Rules(), + start_time=datetime.datetime.now(), + model="gpt-4o-mini", + messages=[{"role": "user", "content": PROMPT}], + litellm_call_id="call-on-arrival", + proxy_server_request={"url": "http://proxy/v1/chat/completions", "headers": dict(TENANT_HEADER), "body": {}}, + metadata={"user_api_key_alias": "audited-key"}, + ) + return {**data, "litellm_logging_obj": logging_obj} + + +async def _pre_call(data: dict, skip_guardrails: bool = False) -> dict: + processed: Final = await ProxyLogging(user_api_key_cache=DualCache()).pre_call_hook( + user_api_key_dict=UserAPIKeyAuth(), data=data, call_type="acompletion", skip_guardrails=skip_guardrails + ) + assert processed is not None + for _ in range(5): + await asyncio.sleep(0) + return processed + + +async def _settle() -> None: + for _ in range(20): + await asyncio.sleep(0) + + +async def _complete( + data: dict, logs: _LogRecorder, mock_response: object = "general kenobi", stream: bool = False +) -> None: + try: + await litellm.acompletion( + model="gpt-4o-mini", + messages=data["messages"], + mock_response=mock_response, + stream=stream, + litellm_logging_obj=data["litellm_logging_obj"], + litellm_call_id=data["litellm_call_id"], + metadata=data["metadata"], + proxy_server_request=data["proxy_server_request"], + ) + except litellm.InternalServerError: + pass + await asyncio.wait_for(logs.logged.wait(), timeout=10) + + +def _vendor_entries(logs: _LogRecorder, event: int = -1) -> list[tuple[object, object]]: + return [ + (entry.get("guardrail_status"), entry.get("guardrail_response")) + for entry in logs.events[event][1] + if entry.get("guardrail_name") == "audit-vendor" + ] + + +@pytest.mark.asyncio +async def test_on_arrival_sends_the_request_with_its_headers_before_the_model_is_called(registered): + guardrail: Final = _on_arrival(_ScanRecorder(gate=asyncio.Event())) + registered.append(guardrail) + + processed: Final = await _pre_call(_proxy_request()) + + assert guardrail.scans == [("request", [PROMPT], TENANT_HEADER)] + assert processed["messages"] == [{"role": "user", "content": PROMPT}], "the scan must never change the request" + + +@pytest.mark.asyncio +async def test_on_arrival_verdict_lands_in_the_success_log_and_the_response_is_not_scanned(registered): + guardrail: Final = _on_arrival(_ScanRecorder()) + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + + data: Final = await _pre_call(_proxy_request()) + assert data["messages"] == [{"role": "user", "content": PROMPT}], "a finished scan must not mask the request" + await _complete(data, logs) + + assert guardrail.scans == [("request", [PROMPT], TENANT_HEADER)] + assert logs.status == "success" + assert [status for status, _ in _vendor_entries(logs)] == ["success"] + + +@pytest.mark.asyncio +async def test_on_arrival_verdict_lands_in_the_failure_log(registered): + guardrail: Final = _on_arrival(_ScanRecorder(block=True)) + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + + await _complete( + await _pre_call(_proxy_request()), + logs, + mock_response=litellm.InternalServerError(message="upstream down", llm_provider="openai", model="gpt-4o-mini"), + ) + + assert logs.status == "failure" + assert [status for status, _ in _vendor_entries(logs)] == ["guardrail_failed_to_respond"] + assert len(guardrail.scans) == 1 + + +@pytest.mark.asyncio +async def test_a_scan_slower_than_the_verdict_timeout_is_cancelled_and_logged_as_not_run(registered): + guardrail: Final = _on_arrival(_ScanRecorder(gate=asyncio.Event()), timeout=0.05) + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + + await _complete(await _pre_call(_proxy_request()), logs) + await asyncio.sleep(0) + + assert _vendor_entries(logs) == [("not_run", "no verdict within 0.05s")] + assert guardrail.cancelled is True + + +@pytest.mark.asyncio +async def test_after_call_timing_sends_a_failed_calls_request_in_the_background_without_a_logged_verdict(registered): + guardrail: Final = _ScanRecorder(gate=asyncio.Event()) + guardrail.logging_only_scope = "input" + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + + await _complete( + await _pre_call(_proxy_request()), + logs, + mock_response=litellm.InternalServerError(message="upstream down", llm_provider="openai", model="gpt-4o-mini"), + ) + await _settle() + + assert logs.status == "failure", "the error is logged while the vendor is still holding the scan" + assert _vendor_entries(logs) == [] + assert guardrail.scans == [("request", [PROMPT], TENANT_HEADER)] + + +@pytest.mark.asyncio +async def test_an_on_arrival_guardrail_falls_back_to_scanning_after_the_call_when_pre_call_was_skipped(registered): + guardrail: Final = _on_arrival(_ScanRecorder()) + logs: Final = _LogRecorder() + registered.extend((guardrail, logs, _PreCallObserver())) + + data: Final = await _pre_call(_proxy_request(), skip_guardrails=True) + assert guardrail.scans == [], "skip_guardrails must not start a scan" + await _complete(data, logs) + + assert guardrail.scans == [("request", [PROMPT], TENANT_HEADER)], "the fallback sends the same headers" + assert [status for status, _ in _vendor_entries(logs)] == ["success"] + + +@pytest.mark.asyncio +async def test_a_guardrail_with_its_own_logging_hook_is_not_run_on_failure(registered): + guardrail: Final = _OwnLoggingHookGuardrail() + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + + await _complete( + await _pre_call(_proxy_request()), + logs, + mock_response=litellm.InternalServerError(message="upstream down", llm_provider="openai", model="gpt-4o-mini"), + ) + await _settle() + + assert guardrail.scans == [] + + +@pytest.mark.asyncio +async def test_verdicts_recorded_before_the_scan_are_not_logged_twice(registered): + guardrail: Final = _on_arrival(_ScanRecorder()) + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + data: Final = _proxy_request() + guardrail.add_standard_logging_guardrail_information_to_request_data( + guardrail_json_response="earlier verdict", request_data=data, guardrail_status="success" + ) + + await _complete(await _pre_call(data), logs) + + assert _vendor_entries(logs).count(("success", "earlier verdict")) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fails", [False, True], ids=["success", "failure"]) +async def test_a_key_opted_out_of_the_guardrail_is_never_scanned(registered, fails): + guardrail: Final = _on_arrival(_ScanRecorder()) + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + data: Final = _proxy_request() + data["metadata"]["user_api_key_metadata"] = {"opted_out_global_guardrails": ["audit-vendor"]} + + await _complete( + await _pre_call(data), + logs, + mock_response=litellm.InternalServerError(message="upstream down", llm_provider="openai", model="gpt-4o-mini") + if fails + else "general kenobi", + ) + await _settle() + + assert guardrail.scans == [] + assert _vendor_entries(logs) == [] + + +@pytest.mark.asyncio +async def test_a_guardrail_without_logging_only_mode_is_not_run_on_failure(registered): + guardrail: Final = _ScanRecorder() + guardrail.event_hook = GuardrailEventHooks.post_call + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + + await _complete( + await _pre_call(_proxy_request()), + logs, + mock_response=litellm.InternalServerError(message="upstream down", llm_provider="openai", model="gpt-4o-mini"), + ) + + assert guardrail.scans == [] + + +@pytest.mark.asyncio +async def test_every_attempt_that_collects_a_timed_out_scan_logs_it_as_not_run(): + guardrail: Final = _on_arrival(_ScanRecorder(gate=asyncio.Event()), timeout=0.05) + scan: Final = asyncio.get_running_loop().create_task( + guardrail.apply_guardrail(inputs={"texts": [PROMPT]}, request_data={"metadata": {}}, input_type="request") + ) + + first: Final = await collect_input_scan(guardrail, scan) + second: Final = await collect_input_scan(guardrail, scan) + + assert [(entry["guardrail_status"], entry["guardrail_response"]) for entry in (*first, *second)] == [ + ("not_run", "no verdict within 0.05s"), + ("not_run", "no verdict within 0.05s"), + ] + + +@pytest.mark.asyncio +async def test_a_request_the_proxy_rejects_before_the_model_is_not_sent_to_the_vendor(registered): + guardrail: Final = _ScanRecorder() + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + logging_obj: Final = _proxy_request()["litellm_logging_obj"] + logging_obj.model_call_details[LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL] = True + + await logging_obj.async_failure_handler( + exception=HTTPException(status_code=401, detail="invalid key"), traceback_exception="" + ) + await _settle() + + assert logs.status == "failure" + assert guardrail.scans == [] + + +@pytest.mark.asyncio +async def test_a_guardrail_scoped_to_non_streaming_calls_skips_a_failed_streaming_call(registered): + guardrail: Final = _ScanRecorder() + guardrail.apply_stream_scope("non_streaming") + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + + await _complete( + await _pre_call(_proxy_request()), + logs, + mock_response=litellm.InternalServerError(message="upstream down", llm_provider="openai", model="gpt-4o-mini"), + stream=True, + ) + await _settle() + + assert guardrail.scans == [] + + +@pytest.mark.asyncio +async def test_a_failed_attempt_leaves_the_scan_running_for_the_retry_that_succeeds(registered): + gate: Final = asyncio.Event() + guardrail: Final = _on_arrival(_ScanRecorder(gate=gate)) + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + data: Final = await _pre_call(_proxy_request()) + + await data["litellm_logging_obj"].async_failure_handler( + exception=litellm.InternalServerError(message="upstream down", llm_provider="openai", model="gpt-4o-mini"), + traceback_exception="", + ) + gate.set() + logs.logged.clear() + await _complete(data, logs) + + assert [status for status, _ in logs.events] == ["failure", "success"] + assert _vendor_entries(logs, event=0) == [("not_run", "no verdict yet when the call failed")] + assert [status for status, _ in _vendor_entries(logs, event=1)] == ["success"] + assert guardrail.cancelled is False + + +@pytest.mark.asyncio +async def test_a_failed_attempts_verdict_reaches_the_payload_the_router_publishes(registered): + guardrail: Final = _on_arrival(_ScanRecorder()) + registered.append(guardrail) + data: Final = await _pre_call(_proxy_request()) + logging_obj: Final = data["litellm_logging_obj"] + + await logging_obj.dispatch_attempt_failure_handlers( + litellm.InternalServerError(message="upstream down", llm_provider="openai", model="gpt-4o-mini"), "" + ) + + published: Final = logging_obj.model_call_details["standard_logging_object"]["guardrail_information"] + assert [entry["guardrail_status"] for entry in published if entry["guardrail_name"] == "audit-vendor"] == [ + "success" + ] + + +@pytest.mark.asyncio +async def test_a_log_masking_guardrail_still_runs_for_a_key_that_disabled_global_guardrails(registered): + guardrail: Final = _OwnLoggingHookGuardrail() + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + data: Final = _proxy_request() + data["metadata"]["user_api_key_metadata"] = {"disable_global_guardrails": True} + + await _complete(await _pre_call(data), logs) + + assert guardrail.scans == [("own-hook", [], None)], "an opted-out key must never turn log redaction off" + + +@pytest.mark.asyncio +async def test_a_call_the_router_rejects_before_routing_is_not_sent_to_the_vendor(registered): + guardrail: Final = _ScanRecorder() + logs: Final = _LogRecorder() + registered.extend((guardrail, logs)) + + await async_log_rejection_before_provider_call( + _proxy_request()["litellm_logging_obj"], + litellm.BadRequestError(message="batch too large", llm_provider="openai", model="gpt-4o-mini"), + ) + await _settle() + + assert logs.status == "failure" + assert guardrail.scans == [] diff --git a/tests/unit/proxy/guardrails/test_guardrail_endpoints.py b/tests/unit/proxy/guardrails/test_guardrail_endpoints.py index 6807a4fb569..7b742ff83f8 100644 --- a/tests/unit/proxy/guardrails/test_guardrail_endpoints.py +++ b/tests/unit/proxy/guardrails/test_guardrail_endpoints.py @@ -1686,6 +1686,37 @@ async def test_patch_guardrail_clears_scope_when_logging_only_mode_is_removed( handler.delete_in_memory_guardrail(stored_guardrail["guardrail_id"]) +@pytest.mark.asyncio +async def test_patch_guardrail_clears_on_arrival_timing_when_logging_only_mode_is_removed( + mocker, monkeypatch, mock_guardrail_registry +): + handler, stored_guardrail = _setup_patch_scope_guardrail( + mocker, + monkeypatch, + mock_guardrail_registry, + _PatchScopeSupportedGuardrail, + "patch_on_arrival_test", + { + "mode": ["pre_call", "logging_only"], + "logging_only_scope": "input", + "logging_only_input_timing": "on_arrival", + "logging_only_verdict_timeout": 5, + "default_on": True, + }, + ) + request = PatchGuardrailRequest(litellm_params=BaseLitellmParams(mode=["pre_call"])) + + try: + await patch_guardrail(stored_guardrail["guardrail_id"], request, user_api_key_dict=MOCK_ADMIN_USER) + + persisted = mock_guardrail_registry.update_guardrail_in_db.call_args.kwargs["guardrail"]["litellm_params"] + assert (persisted.logging_only_input_timing, persisted.logging_only_verdict_timeout) == (None, None) + callback = handler.guardrail_id_to_custom_guardrail[stored_guardrail["guardrail_id"]] + assert callback.logging_only_input_timing == "after_call" + finally: + handler.delete_in_memory_guardrail(stored_guardrail["guardrail_id"]) + + @pytest.mark.asyncio async def test_patch_guardrail_with_blank_value_clears_the_stored_value(mocker, monkeypatch, mock_guardrail_registry): """ diff --git a/tests/unit/proxy/guardrails/test_guardrail_registry.py b/tests/unit/proxy/guardrails/test_guardrail_registry.py index 7eabc24e090..00097051e2c 100644 --- a/tests/unit/proxy/guardrails/test_guardrail_registry.py +++ b/tests/unit/proxy/guardrails/test_guardrail_registry.py @@ -1274,6 +1274,8 @@ class TestLoggingOnlyScopeValidation: reject_invalid_logging_only_scope: bool = False, assert_registered: bool = False, continue_on_input_failure: bool | None = None, + input_timing: str | None = None, + verdict_timeout: float | None = None, ) -> CustomGuardrail: import litellm from litellm.proxy.guardrails import guardrail_registry as registry_module @@ -1308,6 +1310,8 @@ class TestLoggingOnlyScopeValidation: "mode": mode, "logging_only_scope": scope, "logging_only_continue_on_input_failure": continue_on_input_failure, + "logging_only_input_timing": input_timing, + "logging_only_verdict_timeout": verdict_timeout, }, }, reject_invalid_logging_only_scope=reject_invalid_logging_only_scope, @@ -1399,6 +1403,112 @@ class TestLoggingOnlyScopeValidation: assert callback.logging_only_scope == "output" + def test_on_arrival_with_input_scope_applies_timing_and_timeout(self) -> None: + callback: Final = self._initialize( + mode="logging_only", scope="input", input_timing="on_arrival", verdict_timeout=5 + ) + + assert (callback.logging_only_input_timing, callback.logging_only_verdict_timeout) == ("on_arrival", 5) + + def test_on_arrival_without_a_timeout_waits_the_default(self) -> None: + callback: Final = self._initialize(mode="logging_only", scope="input", input_timing="on_arrival") + + assert (callback.logging_only_input_timing, callback.logging_only_verdict_timeout) == ("on_arrival", 3.0) + + @pytest.mark.parametrize( + ("mode", "scope", "input_timing", "verdict_timeout", "callback_type", "message"), + ( + ( + "logging_only", + "output", + "on_arrival", + None, + _LoggingOnlyScopeSupportedGuardrail, + "logging_only_input_timing=on_arrival only applies with logging_only_scope: input. Set " + "logging_only_scope: input or remove logging_only_input_timing.", + ), + ( + "logging_only", + None, + "on_arrival", + None, + _LoggingOnlyScopeSupportedGuardrail, + "logging_only_input_timing=on_arrival only applies with logging_only_scope: input. Set " + "logging_only_scope: input or remove logging_only_input_timing.", + ), + ( + "pre_call", + None, + "on_arrival", + None, + _LoggingOnlyScopeSupportedGuardrail, + "logging_only_input_timing is on_arrival, but mode does not include logging_only, so it would " + "never apply. Add logging_only to mode or remove logging_only_input_timing.", + ), + ( + "logging_only", + "input", + None, + 5, + _LoggingOnlyScopeSupportedGuardrail, + "logging_only_verdict_timeout is set, but logging_only_input_timing is not on_arrival, so it would " + "never apply. Set logging_only_input_timing: on_arrival or remove logging_only_verdict_timeout.", + ), + ( + "logging_only", + None, + "on_arrival", + None, + _LoggingOnlyScopeUnsupportedGuardrail, + "logging_only_input_timing=on_arrival is not supported by this guardrail, whose logging_only hook " + "scans on its own. Remove logging_only_input_timing.", + ), + ), + ids=("output_scope", "both_directions", "mode_without_logging_only", "timeout_alone", "own_logging_hook"), + ) + def test_on_arrival_outside_its_rules_is_ignored_at_load_and_rejected_for_api_writes( + self, + mode: str, + scope: LoggingOnlyScope | None, + input_timing: str | None, + verdict_timeout: float | None, + callback_type: type[CustomGuardrail], + message: str, + ) -> None: + callback: Final = self._initialize( + mode=mode, + scope=scope, + input_timing=input_timing, + verdict_timeout=verdict_timeout, + callback_type=callback_type, + assert_registered=True, + ) + with pytest.raises(ValueError, match="logging_only_") as exc_info: + self._initialize( + mode=mode, + scope=scope, + input_timing=input_timing, + verdict_timeout=verdict_timeout, + callback_type=callback_type, + reject_invalid_logging_only_scope=True, + ) + + assert (callback.logging_only_input_timing, callback.logging_only_verdict_timeout) == ("after_call", 3.0) + assert str(exc_info.value) == f"Guardrail logging-only-scope-guardrail: {message}" + + def test_a_verdict_timeout_below_the_minimum_is_ignored_at_load(self) -> None: + params: Final = { + "guardrail": "litellm_content_filter", + "mode": "logging_only", + "logging_only_scope": "input", + "logging_only_input_timing": "on_arrival", + "logging_only_verdict_timeout": 0.5, + } + + parsed: Final = parse_tolerant_litellm_params(params, "test-content-filter") + + assert (parsed.logging_only_input_timing, parsed.logging_only_verdict_timeout) == ("on_arrival", None) + @pytest.mark.parametrize("scope", ("request", "both")) def test_invalid_scope_fails_litellm_params_validation(self, scope: str) -> None: with pytest.raises(ValidationError): diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 70406105193..c0cf7f6eedd 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -27137,11 +27137,21 @@ export interface components { * @description when True, a flagged or raising logging_only request scan is logged and the response is still scanned, so both verdicts land. Only applies to mode logging_only and is ignored when logging_only_scope is 'input' or 'output'. */ logging_only_continue_on_input_failure?: boolean | null; + /** + * Logging Only Input Timing + * @description when a logging_only request scan runs: 'after_call' (default) scans a copy of the logged request once the call ends; 'on_arrival' sends the request to the guardrail in the background once it passes the pre_call guardrails, in parallel with the model call, and the call's log records the verdict. 'on_arrival' needs mode logging_only, logging_only_scope 'input' and a guardrail without its own logging hook; otherwise API writes are rejected and a config file value is ignored with an error log. Failed model calls are scanned too, without delaying the error or a retry: an 'on_arrival' verdict is logged only if it is already back, and an 'after_call' scan runs in the background without a logged verdict. Requests the proxy rejects before reaching a model (auth, rate limits, budgets, guardrail blocks) are not scanned. + */ + logging_only_input_timing?: ("after_call" | "on_arrival") | null; /** * Logging Only Scope * @description which direction a logging_only scan observes: 'input' (request) or 'output' (response); unset scans both directions. Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking. */ logging_only_scope?: ("input" | "output") | null; + /** + * Logging Only Verdict Timeout + * @description seconds a successful call's log waits for an 'on_arrival' verdict that is not back yet (default 3, minimum 1). When it runs out, the scan is cancelled and logged as not_run. A failed call's log never waits. Only applies with logging_only_input_timing 'on_arrival'. + */ + logging_only_verdict_timeout?: number | null; /** * Mask Request Content * @description Will mask request content if guardrail makes any changes @@ -36870,11 +36880,21 @@ export interface components { * @description when True, a flagged or raising logging_only request scan is logged and the response is still scanned, so both verdicts land. Only applies to mode logging_only and is ignored when logging_only_scope is 'input' or 'output'. */ logging_only_continue_on_input_failure?: boolean | null; + /** + * Logging Only Input Timing + * @description when a logging_only request scan runs: 'after_call' (default) scans a copy of the logged request once the call ends; 'on_arrival' sends the request to the guardrail in the background once it passes the pre_call guardrails, in parallel with the model call, and the call's log records the verdict. 'on_arrival' needs mode logging_only, logging_only_scope 'input' and a guardrail without its own logging hook; otherwise API writes are rejected and a config file value is ignored with an error log. Failed model calls are scanned too, without delaying the error or a retry: an 'on_arrival' verdict is logged only if it is already back, and an 'after_call' scan runs in the background without a logged verdict. Requests the proxy rejects before reaching a model (auth, rate limits, budgets, guardrail blocks) are not scanned. + */ + logging_only_input_timing?: ("after_call" | "on_arrival") | null; /** * Logging Only Scope * @description which direction a logging_only scan observes: 'input' (request) or 'output' (response); unset scans both directions. Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking. */ logging_only_scope?: ("input" | "output") | null; + /** + * Logging Only Verdict Timeout + * @description seconds a successful call's log waits for an 'on_arrival' verdict that is not back yet (default 3, minimum 1). When it runs out, the scan is cancelled and logged as not_run. A failed call's log never waits. Only applies with logging_only_input_timing 'on_arrival'. + */ + logging_only_verdict_timeout?: number | null; /** * Mask * @description Enable content masking using Lasso classifix API