Merge 92822a470b into 5215c2d3f7

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

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:

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)

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
 

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": [
                 {

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(

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)
 

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)
 

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:

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 == []

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):
     """

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):

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