import contextvars
import copy
import hashlib
import os
import secrets
from collections.abc import Mapping
from datetime import datetime
from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, Optional, get_args

from litellm._logging import verbose_logger
from litellm.caching import DualCache
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.core_helpers import (
    get_metadata_variable_name_from_kwargs,
    get_or_create_metadata_bucket,
    redact_nested_match_and_regex_keys,
)
from litellm.secret_managers.main import str_to_bool
from litellm.types.guardrails import (
    DynamicGuardrailParams,
    GuardrailEventHooks,
    LitellmParams,
    Mode,
)
from litellm.types.llms.openai import AllMessageValues
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
from litellm.types.utils import (
    CallTypes,
    GenericGuardrailAPIInputs,
    GuardrailStatus,
    GuardrailTracingDetail,
    LLMResponseTypes,
    StandardLoggingGuardrailInformation,
)

try:
    from fastapi.exceptions import HTTPException
except ImportError:
    HTTPException = None

if TYPE_CHECKING:
    from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
    from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTranslation
dc: Final = DualCache()


from litellm.constants import (
    GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS,
    PRE_CALL_EXECUTED_GUARDRAILS_KEY,
)
from litellm.exceptions import (
    BlockedPiiEntityError,
    GuardrailRaisedException,
    ModifyResponseException,
    SensitiveDataRouteException,
)

# Per-process secret tagging each recorded marker. The deployment hook only
# honors markers carrying this token, so a caller cannot forge the metadata
# field to suppress a guardrail on the direct-SDK path that never reaches the
# proxy's metadata sanitizer.
_PRE_CALL_EXECUTED_TOKEN: Final = secrets.token_hex(16)

_GUARDRAIL_BLOCK_STATUS_CODES: Final = frozenset({400, 403, 422})

DEFAULT_ADVISORY_MESSAGE: Final = (
    "The user's latest message was flagged for {reason} by a content safety "
    "guardrail. This may be a false positive. Use your judgment: respond "
    "helpfully if the request is legitimate, or decline if it is not."
)

_guardrail_self_recorded: Final[contextvars.ContextVar[bool]] = contextvars.ContextVar(
    "litellm_guardrail_self_recorded", default=False
)


def is_guardrail_intervention(e: Exception) -> bool:
    """
    Returns True if the exception represents an intentional guardrail block
    (this was logged previously as an API failure - guardrail_failed_to_respond).

    Guardrails signal intentional blocks by raising:
    - GuardrailRaisedException (generic guardrail API, tool permission)
    - BlockedPiiEntityError (Presidio PII detection)
    - SensitiveDataRouteException (sensitive-data reroute to on-premise model)
    - HTTPException with a block-signalling status (400, 403, 422)
    - ModifyResponseException (passthrough mode violation)

    Only the statuses guardrails use in-tree to signal a deliberate rejection
    count as an intervention: 400 (content policy), 403 (e.g. akto) and 422
    (e.g. llm_as_a_judge). Other 4xx codes are commonly propagated from an
    upstream guardrail provider response (401 bad key, 408 timeout, 429 rate
    limit, or a raw upstream status), which are technical failures, not
    blocks, so they stay guardrail_failed_to_respond.
    """
    if isinstance(e, ModifyResponseException):
        return True
    if isinstance(
        e,
        (
            GuardrailRaisedException,
            BlockedPiiEntityError,
            SensitiveDataRouteException,
        ),
    ):
        return True
    if HTTPException is not None and isinstance(e, HTTPException) and e.status_code in _GUARDRAIL_BLOCK_STATUS_CODES:
        return True
    return False


def _strict_guardrail_modes_enabled() -> bool:
    """Whether guardrail-mode validation raises (default) or logs a warning.

    Set `LITELLM_STRICT_GUARDRAIL_MODES=false` to keep the pre-LIT-4226 behavior
    for guardrails whose supported_event_hooks list newly includes their
    configured mode: log the mismatch and continue instead of raising at boot.
    """
    raw: Final = os.environ.get("LITELLM_STRICT_GUARDRAIL_MODES")
    if raw is None:
        return True
    parsed: Final = str_to_bool(raw)
    return True if parsed is None else parsed


def get_session_id_from_request_data(request_data: dict[str, Any]) -> str | None:
    """Extract session_id from request data (litellm_session_id or metadata)."""
    session_id = request_data.get("litellm_session_id")
    if session_id:
        return str(session_id)

    metadata: Final = request_data.get("metadata") or {}
    session_id = metadata.get("session_id")
    if session_id:
        return str(session_id)

    litellm_metadata: Final = request_data.get("litellm_metadata") or {}
    session_id = litellm_metadata.get("session_id")
    if session_id:
        return str(session_id)

    return None


class CustomGuardrail(CustomLogger):
    # If True, during_call runs async_moderation_hook instead of the unified apply_guardrail path.
    use_native_during_call_hook: ClassVar[bool] = False

    # If True, every proxy lifecycle event runs this guardrail's own hooks, not apply_guardrail.
    use_native_lifecycle_hooks: ClassVar[bool] = False

    records_own_guardrail_information: ClassVar[bool] = False

    def __init__(
        self,
        guardrail_name: str | None = None,
        supported_event_hooks: list[GuardrailEventHooks] | None = None,
        event_hook: GuardrailEventHooks | list[GuardrailEventHooks] | Mode | None = None,
        default_on: bool = False,
        mask_request_content: bool = False,
        mask_response_content: bool = False,
        violation_message_template: str | None = None,
        end_session_after_n_fails: int | None = None,
        on_violation: str | None = None,
        realtime_violation_message: str | None = None,
        on_sensitive_data: str | None = None,
        sensitive_data_route_to_model: str | None = None,
        sticky_session_routing: bool = True,
        run_in_parallel: bool = False,
        scan_raw_request: bool = False,
        only_scan_new_messages: bool = False,
        **kwargs,
    ):
        """
        Initialize the CustomGuardrail class

        Args:
            guardrail_name: The name of the guardrail. This is the name used in your requests.
            supported_event_hooks: The event hooks that the guardrail supports
            event_hook: The event hook to run the guardrail on
            default_on: If True, the guardrail will be run by default on all requests
            mask_request_content: If True, the guardrail will mask the request content
            mask_response_content: If True, the guardrail will mask the response content
            end_session_after_n_fails: For /v1/realtime sessions, end the session after this many violations
            on_violation: For /v1/realtime sessions, 'warn' or 'end_session'
            realtime_violation_message: Message the bot speaks aloud when a /v1/realtime guardrail fires
            on_sensitive_data: Action when sensitive data is detected. 'block' (default) or 'route'
            sensitive_data_route_to_model: Model to route to when on_sensitive_data='route'
            sticky_session_routing: When True, all subsequent requests in the session use the same model
            run_in_parallel: When True, this pre_call or post_call guardrail runs concurrently with
                other opted-in guardrails of the same hook. Only safe for block-only guardrails that
                do not mutate the request or response.
            scan_raw_request: When True, this pre_call guardrail always evaluates the request as it
                was before any guardrail in this hook ran, regardless of where it's declared in the
                guardrails list -- so an earlier guardrail that masks/rewrites content (e.g. PII
                redaction) can never hide a violation from this one. Only safe for block-only
                guardrails: any data this guardrail returns is discarded, matching run_in_parallel's
                contract, since applying its mutations on top of a stale snapshot would silently
                undo whatever later guardrails already did to the live request.
        """
        self.guardrail_name = guardrail_name
        self.supported_event_hooks = supported_event_hooks
        self.event_hook: GuardrailEventHooks | list[GuardrailEventHooks] | Mode | None = event_hook
        self.default_on: bool = default_on
        self.mask_request_content: bool = mask_request_content
        self.mask_response_content: bool = mask_response_content
        self.violation_message_template: str | None = violation_message_template
        self.end_session_after_n_fails: int | None = end_session_after_n_fails
        self.on_violation: str | None = on_violation
        self.realtime_violation_message: str | None = realtime_violation_message
        self.on_sensitive_data: str | None = on_sensitive_data
        self.sensitive_data_route_to_model: str | None = sensitive_data_route_to_model
        self.sticky_session_routing: bool = sticky_session_routing
        self.run_in_parallel: bool = run_in_parallel
        self.scan_raw_request: bool = scan_raw_request
        self.only_scan_new_messages: bool = only_scan_new_messages

        if supported_event_hooks:
            ## validate event_hook is in supported_event_hooks
            try:
                self._validate_event_hook(event_hook, supported_event_hooks)
            except ValueError as validation_error:
                if _strict_guardrail_modes_enabled():
                    raise
                verbose_logger.warning(
                    "%s. LITELLM_STRICT_GUARDRAIL_MODES=false; continuing "
                    "with unsupported event_hook. Set the env var to true "
                    "(default) to enforce validation and fail at startup.",
                    validation_error,
                )
        super().__init__(**kwargs)

    def render_violation_message(self, default: str, context: Mapping[str, object] | None = None) -> str:
        """Return a custom violation message if template is configured."""

        if not self.violation_message_template:
            return default

        format_context: Final[dict[str, object]] = {"default_message": default}
        if context:
            format_context.update(context)
        try:
            return self.violation_message_template.format(**format_context)
        except Exception as e:
            verbose_logger.warning(
                "Failed to format violation message template for guardrail %s: %s",
                self.guardrail_name,
                e,
            )
            return default

    def raise_passthrough_exception(
        self,
        violation_message: str,
        request_data: dict[str, Any],
        detection_info: dict[str, Any] | None = None,
        original_response: object = None,
    ) -> None:
        """
        Raise a passthrough exception for guardrail violations.

        This helper method should be used by guardrails when they detect a violation
        in passthrough mode.

        The exception will be caught by the proxy endpoints and converted to a 200 response
        with the violation message, preventing the LLM call from being made (pre_call/during_call)
        or replacing the LLM response (post_call).

        Args:
            violation_message: The formatted violation message to return to the user
            request_data: The original request data dictionary
            detection_info: Optional dictionary with detection metadata (scores, rules, etc.)
            original_response: The blocked LLM response when raising from a post-call
                hook. It carries the real token usage the upstream call consumed, so
                the synthetic block response reports it instead of zeros. Leave None
                for pre-call/during-call blocks (the LLM was never invoked).

        Raises:
            ModifyResponseException: Always raises this exception to short-circuit
                                     the LLM call and return the violation message

        Example:
            if violation_detected and self.on_flagged_action == "passthrough":
                message = self._format_violation_message(detection_info)
                self.raise_passthrough_exception(
                    violation_message=message,
                    request_data=data,
                    detection_info=detection_info
                )
        """
        model: Final = request_data.get("model", "unknown")

        raise ModifyResponseException(
            message=violation_message,
            model=model,
            request_data=request_data,
            guardrail_name=self.guardrail_name,
            detection_info=detection_info,
            original_response=original_response,
        )

    def inject_advisory_message(
        self,
        data: dict[str, Any],  # mutable-ok: caller's dict is mutated in place, matching mark_pre_call_hook_ran
        message: str,
    ) -> bool:
        """
        Append an advisory system message to the request in place, so the LLM
        itself can weigh a possible false-positive guardrail flag rather than
        the request being hard-blocked or silently allowed.

        Unlike raise_passthrough_exception, this does NOT short-circuit the LLM
        call; the request proceeds normally with the extra message appended.
        Guardrails should call this from on_flagged handling analogous to how
        passthrough-supporting guardrails call raise_passthrough_exception.

        Args:
            data: The request data dictionary, mutated in place to append the
                advisory message to its "messages" list and/or "input"/
                "instructions" text.
            message: The formatted advisory message to append as a system message.

        Returns:
            True if the advisory was actually written somewhere the model will
            see it. False if ``data["input"]`` is a structured Responses-API
            list (not a plain string) -- the Responses API reads only
            ``input``, so appending to ``messages`` would be inert regardless
            of whether a ``messages`` list also happens to be present, and
            there is no field this helper can safely append into. The caller
            must treat this like any other case where the mitigation can't
            land and degrade to blocking instead of silently letting the
            flagged request through unmodified.
        """
        advisory_message: Final = {"role": "system", "content": message}  # mutable-ok: plain dict for live request
        existing_messages: Final = data.get("messages")
        existing_input: Final = data.get("input")
        existing_instructions: Final = data.get("instructions")
        if isinstance(existing_instructions, str):
            # Responses API "instructions" is the privileged, developer-set
            # system-level field the model treats as authoritative -- unlike
            # "input", which the caller controls and could use to tell the
            # model to disregard a trailing warning. Prefer it over "input"
            # whenever present.
            if isinstance(existing_messages, list):
                messages_with_instructions_note: Final = [  # mutable-ok: fresh list
                    *existing_messages,
                    advisory_message,
                ]
                data["messages"] = messages_with_instructions_note  # rebind-ok: mutates caller's dict by design
            data["instructions"] = f"{existing_instructions}\n\n{message}"  # rebind-ok: mutates caller's dict by design
            return True
        if isinstance(existing_input, str):
            # A plain-string "input" doesn't rule out "messages" also being a
            # real, read field (e.g. a chat-completions call carrying a stray
            # "input"), so write to both when both are present.
            if isinstance(existing_messages, list):
                messages_with_input_note: Final = [*existing_messages, advisory_message]  # mutable-ok: fresh list
                data["messages"] = messages_with_input_note  # rebind-ok: mutates caller's dict by design
            # The Responses API reads "input", not "messages" -- appending only to
            # "messages" would leave the advisory unreachable for that endpoint.
            data["input"] = f"{existing_input}\n\n{message}"  # rebind-ok: mutates caller's dict by design
            return True
        if existing_input is not None:
            # existing_input is a structured (non-string) Responses-API item
            # list. That endpoint reads only "input", so appending to
            # "messages" -- even if "messages" also happens to be present --
            # would never reach the model. Leave data untouched and report
            # non-delivery so the caller degrades to blocking.
            return False
        if isinstance(existing_messages, list):
            messages_without_input_note: Final = [*existing_messages, advisory_message]  # mutable-ok: fresh list
            data["messages"] = messages_without_input_note  # rebind-ok: mutates caller's dict by design
            return True
        sole_message: Final = [advisory_message]  # mutable-ok: plain list for the live JSON request
        data["messages"] = sole_message  # rebind-ok: mutates caller's dict by design
        return True

    def raise_sensitive_data_route_exception(
        self,
        route_to_model: str,
        request_data: dict[str, Any],
        detection_info: dict[str, Any] | None = None,
    ) -> None:
        """
        Raise an exception to reroute the request to a different model.

        Use this when sensitive data is detected and the guardrail is configured
        to route to an on-premise model instead of blocking.

        The exception will reroute this request to the specified model. When
        sticky_session_routing is enabled (the default), it also stores the
        routing decision so subsequent requests in this session reuse the model.

        Args:
            route_to_model: The model to route this request (and session) to
            request_data: The original request data dictionary
            detection_info: Optional non-sensitive detection metadata (e.g. matched
                entity types, rule ids, scores). This is surfaced in request metadata
                and logs, so it must not contain the raw detected sensitive values.

        Raises:
            SensitiveDataRouteException: Always raises to trigger rerouting
        """
        session_id: Final = self._get_session_id_from_request_data(request_data)
        if not session_id:
            raise ValueError(
                "Cannot route sensitive data without a session_id. "
                "Ensure the request includes a session_id in metadata or headers."
            )

        raise SensitiveDataRouteException(
            route_to_model=route_to_model,
            session_id=session_id,
            guardrail_name=self.guardrail_name,
            detection_info=detection_info,
            sticky_session_routing=self.sticky_session_routing,
        )

    def _get_session_id_from_request_data(self, request_data: dict[str, Any]) -> str | None:
        """Extract session_id from request data."""
        return get_session_id_from_request_data(request_data)

    @staticmethod
    def _scanned_text_hash(text: str) -> str:
        """Stable content hash for a single scannable text segment.

        Hashing the exact text the provider would receive means an edited earlier
        segment produces a different hash and gets re-scanned, while an unchanged
        segment repeated on a later turn is skipped.
        """
        return hashlib.sha256(text.encode("utf-8")).hexdigest()

    def _scanned_texts_cache_key(self, session_id: str) -> str:
        return f"guardrail_scanned_texts:{self.guardrail_name}:{session_id}"

    async def filter_new_texts_for_session(
        self,
        texts: list[str] | None,
        request_data: dict[str, object],
        cache: DualCache,
    ) -> list[str] | None:
        """Return only the text segments not already scanned earlier in this session.

        Returns ``None`` when incremental scanning is inactive (feature off, no
        session id, masking enabled, or the cache read failed). ``None`` signals
        the caller to fall back to a full scan; a returned list (possibly empty)
        signals the caller to scan only that subset and skip masking write-back.
        """
        if not self.only_scan_new_messages or not texts:
            return None

        if self.mask_request_content or self.mask_response_content:
            verbose_logger.warning(
                "Guardrail %s: only_scan_new_messages is not supported with masking; scanning full context.",
                self.guardrail_name,
            )
            return None

        session_id: Final = get_session_id_from_request_data(request_data)
        if not session_id:
            verbose_logger.debug(
                "Guardrail %s: only_scan_new_messages enabled but request has no session id; scanning full context.",
                self.guardrail_name,
            )
            return None

        try:
            cached: Final[object] = await cache.async_get_cache(key=self._scanned_texts_cache_key(session_id))
        except Exception as e:  # noqa: BLE001  # cache is best-effort; any failure must fall back to a full scan
            verbose_logger.warning(
                "Guardrail %s: failed to read scanned-message cache (%s); scanning full context.",
                self.guardrail_name,
                e,
            )
            return None

        seen: Final[set[str]] = {str(h) for h in cached} if isinstance(cached, list) else set()
        return [text for text in texts if self._scanned_text_hash(text) not in seen]

    async def mark_texts_scanned(
        self,
        texts: list[str] | None,
        request_data: dict[str, object],
        cache: DualCache,
    ) -> None:
        """Record the hashes of all text segments present on a successful (non-blocked) scan.

        Called only after the guardrail allows the request, so a blocked segment is
        never marked scanned and will be re-checked if the client retries.
        """
        if not self.only_scan_new_messages or not texts:
            return
        if self.mask_request_content or self.mask_response_content:
            return
        session_id: Final = get_session_id_from_request_data(request_data)
        if not session_id:
            return

        cache_key: Final = self._scanned_texts_cache_key(session_id)
        current_hashes: Final = [self._scanned_text_hash(text) for text in texts]
        try:
            existing: Final[object] = await cache.async_get_cache(key=cache_key)
            existing_hashes: Final[list[str]] = [str(h) for h in existing] if isinstance(existing, list) else []
            merged: Final[list[str]] = list(dict.fromkeys(existing_hashes + current_hashes))
            await cache.async_set_cache(
                key=cache_key,
                value=merged,
                ttl=GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS,
            )
        except Exception as e:  # noqa: BLE001  # cache is best-effort; any failure must not block the request
            verbose_logger.warning(
                "Guardrail %s: failed to persist scanned-message cache (%s); next call will re-scan.",
                self.guardrail_name,
                e,
            )

    def should_route_on_sensitive_data(self) -> bool:
        """
        Returns True if this guardrail is configured to route requests
        to a different model when sensitive data is detected.
        """
        return self.on_sensitive_data == "route" and self.sensitive_data_route_to_model is not None

    def handle_sensitive_data_detection(
        self,
        request_data: dict[str, Any],
        detection_info: dict[str, Any] | None = None,
    ) -> None:
        """
        Handle sensitive data detection based on guardrail configuration.

        If on_sensitive_data='route', raises SensitiveDataRouteException to reroute.
        Otherwise, raises GuardrailRaisedException to block. When routing is
        configured but the request carries no session_id, routing is not possible
        so the request falls back to a graceful block.

        Args:
            request_data: The request data dictionary
            detection_info: Optional non-sensitive detection metadata. When routing,
                this is surfaced in request metadata and logs, so it must not contain
                the raw detected sensitive values.

        Raises:
            SensitiveDataRouteException: When configured to route and a session_id is present
            GuardrailRaisedException: When configured to block, or when routing is
                configured but no session_id is available
        """
        if self.should_route_on_sensitive_data():
            try:
                self.raise_sensitive_data_route_exception(
                    route_to_model=self.sensitive_data_route_to_model,
                    request_data=request_data,
                    detection_info=detection_info,
                )
            except ValueError:
                raise GuardrailRaisedException(
                    message=(
                        f"Sensitive data detected by {self.guardrail_name} (routing skipped: request has no session_id)"
                    ),
                    guardrail_name=self.guardrail_name,
                    blocked_content=True,
                )
        else:
            raise GuardrailRaisedException(
                message=f"Sensitive data detected by {self.guardrail_name}",
                guardrail_name=self.guardrail_name,
                blocked_content=True,
            )

    @staticmethod
    def get_config_model() -> type["GuardrailConfigModel"] | None:
        """
        Returns the config model for the guardrail

        This is used to render the config model in the UI.
        """
        return None

    @classmethod
    def get_supported_event_hooks(cls) -> list[GuardrailEventHooks] | None:
        """
        Returns the event hooks this guardrail supports, for the UI to render.

        Subclasses should override to return their supported hooks list. When a
        subclass returns None, the endpoint omits it from the per-provider map
        and the UI is expected to fall back to the global `supported_modes`
        list client-side.
        """
        return None

    def _validate_event_hook(
        self,
        event_hook: GuardrailEventHooks | list[GuardrailEventHooks] | Mode | None,
        supported_event_hooks: list[GuardrailEventHooks],
    ) -> None:
        def _validate_event_hook_list_is_in_supported_event_hooks(
            event_hook: list[GuardrailEventHooks] | list[str],
            supported_event_hooks: list[GuardrailEventHooks],
        ) -> None:
            for hook in event_hook:
                if isinstance(hook, str):
                    hook = GuardrailEventHooks(hook)
                if hook not in supported_event_hooks:
                    raise ValueError(f"Event hook {hook} is not in the supported event hooks {supported_event_hooks}")

        if event_hook is None:
            return
        if isinstance(event_hook, str):
            event_hook = GuardrailEventHooks(event_hook)
        if isinstance(event_hook, list):
            _validate_event_hook_list_is_in_supported_event_hooks(event_hook, supported_event_hooks)
        elif isinstance(event_hook, Mode):
            tag_values_flat: Final[list] = []
            for v in event_hook.tags.values():
                if isinstance(v, list):
                    tag_values_flat.extend(v)
                else:
                    tag_values_flat.append(v)
            _validate_event_hook_list_is_in_supported_event_hooks(tag_values_flat, supported_event_hooks)
            if event_hook.default:
                default_list = event_hook.default if isinstance(event_hook.default, list) else [event_hook.default]
                _validate_event_hook_list_is_in_supported_event_hooks(default_list, supported_event_hooks)
        elif isinstance(event_hook, GuardrailEventHooks):
            if event_hook not in supported_event_hooks:
                raise ValueError(f"Event hook {event_hook} is not in the supported event hooks {supported_event_hooks}")

    @staticmethod
    def _get_admin_metadata(data: dict) -> dict:
        """Return merged admin-configured key and team metadata from the request data.

        The proxy may inject admin metadata (user_api_key_metadata,
        user_api_key_team_metadata) into either ``metadata`` or
        ``litellm_metadata`` depending on endpoint. Check both so a caller
        cannot shadow admin config by pre-populating the other key.
        Key-level settings override team-level.
        """
        team_meta: dict = {}
        key_meta: dict = {}
        for key in ("metadata", "litellm_metadata"):
            # Defensive: an unparsed JSON-string metadata could leak past the
            # proxy's normal parse path; don't AttributeError on .get().
            meta = data.get(key)
            if not isinstance(meta, dict):
                continue
            team_meta = meta.get("user_api_key_team_metadata") or team_meta
            key_meta = meta.get("user_api_key_metadata") or key_meta
        return {**team_meta, **key_meta}

    def get_disable_global_guardrail(self, data: dict) -> bool | None:
        """
        Returns True if the global guardrail should be disabled.

        Reads from admin-configured key/team metadata only, not from
        the request body, to prevent callers from disabling guardrails.
        """
        return self._get_admin_metadata(data).get("disable_global_guardrails", False)

    def get_opted_out_global_guardrails_from_metadata(self, data: dict) -> list[str]:
        """
        Returns the list of global guardrail names the team/key has opted out of.

        Reads from admin-configured key/team metadata only.
        """
        value: Final = self._get_admin_metadata(data).get("opted_out_global_guardrails")
        return value if isinstance(value, list) else []

    def _is_valid_response_type(self, result: object) -> bool:
        """
        Check if result is a valid LLMResponseTypes instance.

        Safely handles TypedDict types which don't support isinstance checks.
        For non-LiteLLM responses (like passthrough httpx.Response), returns True
        to allow them through.
        """
        if result is None:
            return False

        try:
            # Try isinstance check on valid types that support it
            response_types: Final = get_args(LLMResponseTypes)
            return isinstance(result, response_types)
        except TypeError as e:
            # TypedDict types don't support isinstance checks
            # In this case, we can't validate the type, so we allow it through
            if "TypedDict" in str(e):
                return True
            raise

    def get_guardrail_from_metadata(self, data: dict) -> list[str] | list[dict[str, DynamicGuardrailParams]]:
        """
        Returns the guardrail(s) to be run from the metadata or root
        """

        if "guardrails" in data:
            return data["guardrails"]
        # Check both metadata locations. For regular endpoints move_guardrails_to_metadata
        # writes to "metadata"; for thread/assistant endpoints it writes to
        # "litellm_metadata". We check the one that actually contains the "guardrails"
        # key so that a non-empty litellm_metadata without guardrails does not shadow
        # the merged list stored in metadata (which would cause team guardrails to be
        # silently skipped while default_on=True policy guardrails still fire).
        for meta_key in ("metadata", "litellm_metadata"):
            meta = data.get(meta_key) or {}
            if isinstance(meta, dict) and "guardrails" in meta:
                return meta.get("guardrails") or []
        return []

    def _guardrail_is_in_requested_guardrails(
        self,
        requested_guardrails: list[str] | list[dict[str, DynamicGuardrailParams]],
    ) -> bool:
        for _guardrail in requested_guardrails:
            if isinstance(_guardrail, dict):
                if self.guardrail_name in _guardrail:
                    return True
            elif isinstance(_guardrail, str):
                if self.guardrail_name == _guardrail:
                    return True

        return False

    def _pre_call_marker(self) -> str | None:
        name: Final = self.guardrail_name
        if not name:
            return None
        return f"{_PRE_CALL_EXECUTED_TOKEN}:{name}"

    def mark_pre_call_hook_ran(self, data: dict[str, object]) -> None:
        """
        Record that this guardrail's ``async_pre_call_hook`` already ran for this
        request, so the deployment-level hook does not run it a second time.

        The proxy runs pre-call guardrails in ``ProxyLogging.pre_call_hook``. The
        router later spreads a deployment's model-level ``guardrails`` into the
        top-level request kwargs, which would otherwise re-trigger the same hook
        from ``async_pre_call_deployment_hook``.
        """
        marker: Final = self._pre_call_marker()
        if marker is None:
            return
        for meta_key in ("metadata", "litellm_metadata"):
            meta = data.get(meta_key)
            if isinstance(meta, dict):
                executed = meta.get(PRE_CALL_EXECUTED_GUARDRAILS_KEY)
                if isinstance(executed, list):
                    if marker not in executed:
                        executed.append(marker)
                else:
                    meta[PRE_CALL_EXECUTED_GUARDRAILS_KEY] = [marker]
                return
        data["metadata"] = {PRE_CALL_EXECUTED_GUARDRAILS_KEY: [marker]}

    def _pre_call_hook_already_ran(self, data: dict[str, object]) -> bool:
        marker: Final = self._pre_call_marker()
        if marker is None:
            return False
        for meta_key in ("metadata", "litellm_metadata"):
            meta = data.get(meta_key)
            if isinstance(meta, dict):
                executed = meta.get(PRE_CALL_EXECUTED_GUARDRAILS_KEY)
                if isinstance(executed, list) and marker in executed:
                    return True
        return False

    def uses_apply_guardrail_interface(self) -> bool:
        return type(self).apply_guardrail is not CustomGuardrail.apply_guardrail

    def _deployment_pre_call_target(self) -> "CustomLogger":
        if not self.uses_apply_guardrail_interface() or self.use_native_lifecycle_hooks:
            return self
        try:
            from litellm.proxy.utils import unified_guardrail
        except ImportError as e:
            raise ImportError(
                f"Guardrail {self.guardrail_name or type(self).__name__} implements apply_guardrail, which needs "
                "the litellm proxy dependencies to run at the deployment level. "
                "Install them with: pip install 'litellm[proxy]'"
            ) from e
        return unified_guardrail

    async def async_pre_call_deployment_hook(self, kwargs: dict[str, Any], call_type: CallTypes | None) -> dict | None:
        from litellm.proxy._types import UserAPIKeyAuth

        # should run guardrail
        litellm_guardrails: Final = kwargs.get("guardrails")
        if litellm_guardrails is None or not isinstance(litellm_guardrails, list):
            return kwargs

        if self._pre_call_hook_already_ran(kwargs):
            return kwargs

        if self.should_run_guardrail(data=kwargs, event_type=GuardrailEventHooks.pre_call) is not True:
            return kwargs

        # CHECK IF GUARDRAIL REJECTS THE REQUEST
        if call_type == CallTypes.completion or call_type == CallTypes.acompletion:
            target: Final = self._deployment_pre_call_target()
            if target is not self:
                kwargs["guardrail_to_apply"] = self
            result: Final = await target.async_pre_call_hook(
                user_api_key_dict=UserAPIKeyAuth(
                    user_id=kwargs.get("user_api_key_user_id"),
                    team_id=kwargs.get("user_api_key_team_id"),
                    end_user_id=kwargs.get("user_api_key_end_user_id"),
                    api_key=kwargs.get("user_api_key_hash"),
                    request_route=kwargs.get("user_api_key_request_route"),
                ),
                cache=dc,
                data=kwargs,
                call_type="completion" if call_type == CallTypes.completion else "acompletion",
            )

            if result is not None and isinstance(result, dict):
                result_messages: Final = result.get("messages")
                if result_messages is not None:  # update for any pii / masking logic
                    kwargs["messages"] = result_messages

        return kwargs

    async def async_post_call_success_deployment_hook(
        self,
        request_data: dict,
        response: LLMResponseTypes,
        call_type: CallTypes | None,
    ) -> LLMResponseTypes | None:
        """
        Allow modifying / reviewing the response just after it's received from the deployment.
        """
        from litellm.proxy._types import UserAPIKeyAuth

        # should run guardrail
        litellm_guardrails: Final = request_data.get("guardrails")
        if litellm_guardrails is None or not isinstance(litellm_guardrails, list):
            return response

        if self.should_run_guardrail(data=request_data, event_type=GuardrailEventHooks.post_call) is not True:
            return response

        # CHECK IF GUARDRAIL REJECTS THE REQUEST
        result: Final = await self.async_post_call_success_hook(
            user_api_key_dict=UserAPIKeyAuth(
                user_id=request_data.get("user_api_key_user_id"),
                team_id=request_data.get("user_api_key_team_id"),
                end_user_id=request_data.get("user_api_key_end_user_id"),
                api_key=request_data.get("user_api_key_hash"),
                request_route=request_data.get("user_api_key_request_route"),
            ),
            data=request_data,
            response=response,
        )

        if not self._is_valid_response_type(result):
            return response

        return result

    async def async_logging_hook(
        self,
        kwargs: dict,  # mutable-ok: CustomLogger.async_logging_hook contract
        result: object,
        call_type: str,
    ) -> tuple[dict, object]:  # mutable-ok: CustomLogger.async_logging_hook contract
        """logging_only: run apply_guardrail on copies of the logged request/response and record the verdict."""
        from litellm.llms import get_guardrail_translation_mapping

        if not self.uses_apply_guardrail_interface() or self.use_native_lifecycle_hooks:
            return kwargs, result
        try:
            translation: Final = get_guardrail_translation_mapping(CallTypes(call_type))()
        except ValueError:
            verbose_logger.debug(
                "Guardrail %s: no guardrail translation for call_type=%s, skipping logging_only scan",
                self.guardrail_name,
                call_type,
            )
            return kwargs, result
        litellm_params: Final = kwargs.get("litellm_params") or {}
        scratch_metadata: Final = {
            key: value
            for key, value in (litellm_params.get("metadata") or {}).items()
            if key != "standard_logging_guardrail_information"
        }
        try:
            await self._scan_logged_call(kwargs, result, 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

    async def _scan_logged_call(
        self,
        kwargs: dict,  # mutable-ok: CustomLogger.async_logging_hook contract
        result: object,
        translation: "BaseTranslation",
        scratch_metadata: dict,  # mutable-ok: apply_guardrail records its verdict into request metadata
    ) -> None:
        optional_params: Final = kwargs.get("optional_params") or {}
        scratch_input: Final = copy.deepcopy(kwargs.get("messages") or kwargs.get("input"))
        scratch_request: Final = {
            "model": kwargs.get("model"),
            "messages": scratch_input,
            "input": scratch_input,
            "tools": copy.deepcopy(optional_params.get("tools")),
            "litellm_call_id": kwargs.get("litellm_call_id"),
            "metadata": scratch_metadata,
        }
        await translation.process_input_messages(data=scratch_request, guardrail_to_apply=self)
        await translation.process_output_response(
            response=copy.deepcopy(result), guardrail_to_apply=self, request_data=scratch_request
        )

    def supports_scan_only_tool_results(self) -> bool:
        """Whether this guardrail can scan tool-result content.

        Guardrails whose own role filtering only ever scans human-authored
        messages override this to return False, so configuring them with
        ``scan_only_tool_results`` is rejected at initialization instead of
        silently scanning nothing on every request.
        """
        return True

    def structured_messages_cover_full_request(self) -> bool:
        """Whether returned ``structured_messages`` span the whole request.

        Translation handlers hand guardrails only the in-scope subset of the
        conversation and merge a returned ``structured_messages`` list back
        into the full request. A guardrail that already rebuilds the complete
        conversation itself (like CrowdStrike AIDR with its skip filters
        active) overrides this to return True so the handler installs the
        returned list as-is instead of merging it a second time, which would
        duplicate the out-of-scope messages.
        """
        return False

    def should_run_guardrail(
        self,
        data,
        event_type: GuardrailEventHooks,
    ) -> bool:
        """
        Returns True if the guardrail should be run on the event_type
        """
        requested_guardrails: Final = self.get_guardrail_from_metadata(data)
        disable_global_guardrail: Final = self.get_disable_global_guardrail(data)
        opted_out_global_guardrails: Final = self.get_opted_out_global_guardrails_from_metadata(data)
        verbose_logger.debug(
            "inside should_run_guardrail for guardrail=%s event_type= %s guardrail_supported_event_hooks= %s requested_guardrails= %s self.default_on= %s",
            self.guardrail_name,
            event_type,
            self.event_hook,
            requested_guardrails,
            self.default_on,
        )
        if self.default_on is True and self.guardrail_name in opted_out_global_guardrails:
            return False

        if self.default_on is True and disable_global_guardrail is True:
            return False

        if self.default_on is True and disable_global_guardrail is not True:
            if self._event_hook_is_event_type(event_type):
                if isinstance(self.event_hook, Mode):
                    try:
                        from litellm_enterprise.integrations.custom_guardrail import (
                            EnterpriseCustomGuardrailHelper,
                        )
                    except ImportError:
                        raise ImportError(
                            "Setting tag-based guardrails is only available in litellm-enterprise. You must be a premium user to use this feature."
                        )
                    result = EnterpriseCustomGuardrailHelper._should_run_if_mode_by_tag(
                        data, self.event_hook, event_type
                    )
                    if result is not None:
                        return result
                return True
            return False

        if (
            self.event_hook
            and not self._guardrail_is_in_requested_guardrails(requested_guardrails)
            and event_type.value != "logging_only"
        ):
            return False

        if not self._event_hook_is_event_type(event_type):
            return False

        if isinstance(self.event_hook, Mode):
            try:
                from litellm_enterprise.integrations.custom_guardrail import (
                    EnterpriseCustomGuardrailHelper,
                )
            except ImportError:
                raise ImportError(
                    "Setting tag-based guardrails is only available in litellm-enterprise. You must be a premium user to use this feature."
                )
            result = EnterpriseCustomGuardrailHelper._should_run_if_mode_by_tag(data, self.event_hook, event_type)
            if result is not None:
                return result
        return True

    def _event_hook_is_event_type(self, event_type: GuardrailEventHooks) -> bool:
        """
        Returns True if the event_hook is the same as the event_type

        eg. if `self.event_hook == "pre_call" and event_type == "pre_call"` -> then True
        eg. if `self.event_hook == "pre_call" and event_type == "post_call"` -> then False
        """

        if self.event_hook is None:
            return True
        if isinstance(self.event_hook, list):
            return event_type.value in self.event_hook
        if isinstance(self.event_hook, Mode):
            for tag_value in self.event_hook.tags.values():
                if isinstance(tag_value, list):
                    if event_type.value in tag_value:
                        return True
                elif event_type.value == tag_value:
                    return True
            if self.event_hook.default:
                default_list: Final = (
                    self.event_hook.default if isinstance(self.event_hook.default, list) else [self.event_hook.default]
                )
                return event_type.value in default_list
            return False
        return self.event_hook == event_type.value

    def get_guardrail_dynamic_request_body_params(self, request_data: dict) -> dict:
        """
        Returns `extra_body` to be added to the request body for the Guardrail API call

        Use this to pass dynamic params to the guardrail API call - eg. success_threshold, failure_threshold, etc.

        ```
        [{"lakera_guard": {"extra_body": {"foo": "bar"}}}]
        ```

        Will return: for guardrail=`lakera-guard`:
        {
            "foo": "bar"
        }

        Args:
            request_data: The original `request_data` passed to LiteLLM Proxy
        """
        requested_guardrails: Final = self.get_guardrail_from_metadata(request_data)

        # Look for the guardrail configuration matching self.guardrail_name
        for guardrail in requested_guardrails:
            if isinstance(guardrail, dict) and self.guardrail_name in guardrail:
                # Get the configuration for this guardrail
                guardrail_config: DynamicGuardrailParams = DynamicGuardrailParams(**guardrail[self.guardrail_name])
                extra_body = guardrail_config.get("extra_body", {})
                if self._validate_premium_user() is not True:
                    if isinstance(extra_body, dict) and extra_body:
                        verbose_logger.warning(
                            "Guardrail %s: ignoring dynamic extra_body keys %s because premium_user is False",
                            self.guardrail_name,
                            list(extra_body.keys()),
                        )
                    return {}

                # Return the extra_body if it exists, otherwise empty dict
                return extra_body

        return {}

    def _validate_premium_user(self) -> bool:
        """
        Returns True if the user is a premium user
        """
        from litellm.proxy.proxy_server import CommonProxyErrors, premium_user

        if premium_user is not True:
            verbose_logger.warning(
                "Trying to use premium guardrail without premium user %s", CommonProxyErrors.not_premium_user.value
            )
            return False
        return True

    def add_standard_logging_guardrail_information_to_request_data(
        self,
        guardrail_json_response: Exception | str | dict | list[dict],
        request_data: dict,
        guardrail_status: GuardrailStatus,
        start_time: float | None = None,
        end_time: float | None = None,
        duration: float | None = None,
        masked_entity_count: dict[str, int] | None = None,
        guardrail_provider: str | None = None,
        event_type: GuardrailEventHooks | None = None,
        tracing_detail: GuardrailTracingDetail | None = None,
    ) -> None:
        """
        Builds `StandardLoggingGuardrailInformation` and adds it to the request metadata so it can be used for logging to DataDog, Langfuse, etc.

        Args:
            tracing_detail: Optional typed dict with provider-specific tracing fields
                (guardrail_id, policy_template, detection_method, confidence_score,
                classification, match_details, patterns_checked, alert_recipients).
        """
        if isinstance(guardrail_json_response, Exception):
            guardrail_json_response = str(guardrail_json_response)
        from litellm.types.utils import GuardrailMode

        # Use event_type if provided, otherwise fall back to self.event_hook
        guardrail_mode: GuardrailEventHooks | GuardrailMode | list[GuardrailEventHooks]
        if event_type is not None:
            guardrail_mode = event_type
        elif isinstance(self.event_hook, Mode):
            guardrail_mode = GuardrailMode(**dict(self.event_hook.model_dump()))
        else:
            guardrail_mode = self.event_hook

        from litellm.litellm_core_utils.core_helpers import (
            filter_exceptions_from_params,
        )

        # Sanitize the response to ensure it's JSON serializable and free of circular refs
        # This prevents RecursionErrors in downstream loggers (Langfuse, Datadog, etc.)
        clean_guardrail_response = filter_exceptions_from_params(guardrail_json_response)

        # Strip secret_fields to prevent plaintext Authorization headers from
        # being persisted to spend logs, OTEL traces, or other logging backends.
        # This matches the pattern used by Langfuse and Arize integrations.
        if isinstance(clean_guardrail_response, dict):
            clean_guardrail_response.pop("secret_fields", None)
        elif isinstance(clean_guardrail_response, list):
            for item in clean_guardrail_response:
                if isinstance(item, dict):
                    item.pop("secret_fields", None)

        # Default-safe behavior: never persist raw matched spans in standard
        # guardrail logging payloads (single shared implementation; Bedrock hooks pass
        # raw provider JSON so redaction is not duplicated upstream).
        clean_guardrail_response = redact_nested_match_and_regex_keys(clean_guardrail_response)

        from litellm.litellm_core_utils.sensitive_data_masker import (
            mask_credentials_in_payload,
        )

        clean_guardrail_response = mask_credentials_in_payload(clean_guardrail_response)

        slg: Final = StandardLoggingGuardrailInformation(
            guardrail_name=self.guardrail_name,
            guardrail_provider=guardrail_provider,
            guardrail_mode=guardrail_mode,
            guardrail_response=clean_guardrail_response,
            guardrail_status=guardrail_status,
            start_time=start_time,
            end_time=end_time,
            duration=duration,
            masked_entity_count=masked_entity_count,
            **(tracing_detail or {}),
        )

        def _append_guardrail_info(container: dict) -> None:
            key: Final = "standard_logging_guardrail_information"
            existing: Final = container.get(key)
            if existing is None:
                container[key] = [slg]
            elif isinstance(existing, list):
                existing.append(slg)
            else:
                # should not happen
                container[key] = [existing, slg]

        _, metadata_bucket = get_or_create_metadata_bucket(request_data)
        _append_guardrail_info(metadata_bucket)

        _guardrail_self_recorded.set(True)

        # Emit the otel guardrail span here, where every guardrail execution lands,
        # rather than relying on a post-call hook that does not fire on every path
        # (e.g. a pass-through request that passes its guardrails).
        try:
            from litellm.integrations.otel.logger import emit_guardrail_span

            emit_guardrail_span(slg)
        except Exception:
            pass

    async def apply_guardrail(
        self,
        inputs: GenericGuardrailAPIInputs,
        request_data: dict,
        input_type: Literal["request", "response"],
        logging_obj: Optional["LiteLLMLoggingObj"] = None,
    ) -> GenericGuardrailAPIInputs:
        """
        Apply your guardrail logic to the given inputs

        Args:
            inputs: Dictionary containing:
                - texts: List of texts to apply the guardrail to
                - images: Optional list of images to apply the guardrail to
                - tool_calls: Optional list of tool calls to apply the guardrail to
            request_data: The request data dictionary - containing user api key metadata (e.g. user_id, team_id, etc.)
            input_type: The type of input to apply the guardrail to - "request" or "response"
            logging_obj: Optional logging object for tracking the guardrail execution

        Any of the custom guardrails can override this method to provide custom guardrail logic

        Returns the texts with the guardrail applied and the images with the guardrail applied (if any)

        Raises:
            Exception:
                - If the guardrail raises an exception

        """
        return inputs

    def _process_response(
        self,
        response: dict | None,
        request_data: dict,
        start_time: float | None = None,
        end_time: float | None = None,
        duration: float | None = None,
        event_type: GuardrailEventHooks | None = None,
        original_inputs: dict | None = None,
    ):
        """
        Add StandardLoggingGuardrailInformation to the request data

        This gets logged on downsteam Langfuse, DataDog, etc.
        """
        # Convert None to empty dict to satisfy type requirements
        guardrail_response: dict[str, object] | str = {} if response is None else response

        # For apply_guardrail functions in custom_code_guardrail scenario,
        # simplify the logged response to "allow", "deny", or "mask"
        if original_inputs is not None and isinstance(response, dict):
            # Check if inputs were modified by comparing them
            if self._inputs_were_modified(original_inputs, response):
                guardrail_response = "mask"
            else:
                guardrail_response = "allow"

        verbose_logger.debug("Guardrail response: %s", response)

        self.add_standard_logging_guardrail_information_to_request_data(
            guardrail_json_response=guardrail_response,
            request_data=request_data,
            guardrail_status="success",
            duration=duration,
            start_time=start_time,
            end_time=end_time,
            event_type=event_type,
        )
        return response

    @staticmethod
    def _is_guardrail_intervention(e: Exception) -> bool:
        """Retained spelling for existing callers; prefer ``is_guardrail_intervention``."""
        return is_guardrail_intervention(e)

    def _process_error(
        self,
        e: Exception,
        request_data: dict,
        start_time: float | None = None,
        end_time: float | None = None,
        duration: float | None = None,
        event_type: GuardrailEventHooks | None = None,
    ):
        """
        Add StandardLoggingGuardrailInformation to the request data

        This gets logged on downsteam Langfuse, DataDog, etc.
        """
        guardrail_status: Final[GuardrailStatus] = (
            "guardrail_intervened" if self._is_guardrail_intervention(e) else "guardrail_failed_to_respond"
        )
        # For custom_code_guardrail scenario, log as "deny" instead of full exception
        # Check if this is from custom_code_guardrail by checking the class name
        guardrail_response: Exception | str = e
        if "CustomCodeGuardrail" in self.__class__.__name__:
            guardrail_response = "deny"

        self.add_standard_logging_guardrail_information_to_request_data(
            guardrail_json_response=guardrail_response,
            request_data=request_data,
            guardrail_status=guardrail_status,
            duration=duration,
            start_time=start_time,
            end_time=end_time,
            event_type=event_type,
        )
        raise e

    def _inputs_were_modified(self, original_inputs: dict, response: dict) -> bool:
        """
        Compare original inputs with response to determine if content was modified.

        Returns True if the inputs were modified (mask scenario), False otherwise (allow scenario).
        """
        # Get all keys from both dictionaries
        all_keys: Final = set(original_inputs.keys()) | set(response.keys())

        # Compare each key's value
        for key in all_keys:
            original_value = original_inputs.get(key)
            response_value = response.get(key)
            if original_value != response_value:
                return True

        # No modifications detected
        return False

    def mask_content_in_string(
        self,
        content_string: str,
        mask_string: str,
        start_index: int,
        end_index: int,
    ) -> str:
        """
        Mask the content in the string between the start and end indices.
        """

        # Do nothing if the start or end are not valid
        if not (0 <= start_index < end_index <= len(content_string)):
            return content_string

        # Mask the content
        return content_string[:start_index] + mask_string + content_string[end_index:]

    def update_in_memory_litellm_params(self, litellm_params: LitellmParams) -> None:
        """
        Update the guardrails litellm params in memory
        """
        for key, value in vars(litellm_params).items():
            setattr(self, key, value)

    def get_guardrails_messages_for_call_type(
        self, call_type: CallTypes, data: dict | None = None
    ) -> list[AllMessageValues] | None:
        """
        Returns the messages for the given call type and data
        """
        if call_type is None or data is None:
            return None

        #########################################################
        # /chat/completions
        # /messages
        # Both endpoints store the messages in the "messages" key
        #########################################################
        if (
            call_type == CallTypes.completion.value
            or call_type == CallTypes.acompletion.value
            or call_type == CallTypes.anthropic_messages.value
            or call_type == CallTypes.call_mcp_tool.value
        ):
            return data.get("messages")

        #########################################################
        # /responses
        # User/System messages are stored in the "input" key, use litellm transformation to get the messages
        #########################################################
        if call_type == CallTypes.responses.value or call_type == CallTypes.aresponses.value:
            from typing import cast

            from litellm.responses.litellm_completion_transformation.transformation import (
                LiteLLMCompletionResponsesConfig,
            )

            input_data: Final = data.get("input")
            if input_data is None:
                return None

            messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
                input=input_data,
                responses_api_request=data,
            )
            return cast(list[AllMessageValues], messages)
        return None


def _append_slg_to_litellm_params(lp: object, entries: list) -> None:
    """Merge guardrail entries into a single litellm_params dict."""
    if not isinstance(lp, dict):
        return
    if lp.get("metadata") is None:
        lp["metadata"] = {}
    existing: Final = lp["metadata"].setdefault("standard_logging_guardrail_information", [])
    for entry in entries:
        if entry not in existing:
            existing.append(entry)


def _sync_guardrail_info_to_logging_obj(request_data: dict, logging_obj: object) -> None:
    """Copy standard_logging_guardrail_information from request_data into logging_obj.

    The @log_guardrail_information decorator writes guardrail info to
    request_data["metadata"] or request_data["litellm_metadata"].  For
    passthrough routes (/v1/messages, /v1/responses) the spend-log payload is
    built from logging_obj.litellm_params["metadata"], which is a separate dict
    that does not share identity with the one in request_data.  This helper
    bridges that gap so guardrail_information is non-null in spend logs for all
    routes, not just /v1/chat/completions.
    """
    if logging_obj is None:
        return
    meta_src: Final = request_data.get(get_metadata_variable_name_from_kwargs(request_data)) or {}
    slg_info: Final = meta_src.get("standard_logging_guardrail_information")
    if not slg_info:
        return
    entries: Final[list] = slg_info if isinstance(slg_info, list) else [slg_info]
    mcd: Final = getattr(logging_obj, "model_call_details", None) or {}
    _append_slg_to_litellm_params(getattr(logging_obj, "litellm_params", None), entries)
    _append_slg_to_litellm_params(mcd.get("litellm_params"), entries)


def log_guardrail_information(func):
    """
    Decorator to add standard logging guardrail information to any function

    Add this decorator to ensure your guardrail response is logged to DataDog, OTEL, s3, GCS etc.

    Logs for:
        - pre_call
        - during_call
        - post_call

    Some guardrails (e.g. ``block_code_execution``) call
    ``add_standard_logging_guardrail_information_to_request_data`` directly
    from inside the wrapped function so they can record a richer payload
    (structured detections, tracing detail) than this decorator's
    "allow"/"mask"/raw-response default. To avoid double-recording in that
    case (which would emit two spans, two Datadog records, two spend-log
    entries, etc.), a context-local flag records whether the wrapped function
    appended its own entry; if so, the auto-record is skipped. The flag is a
    ``ContextVar`` rather than a count of entries in the shared ``request_data``
    so it stays correct when guardrails run concurrently (asyncio copies the
    context into each gathered task): counting shared entries would let one
    guardrail's append hide another guardrail's missing record.

    A guardrail that only records an entry when it actually runs (e.g.
    ``HeadroomGuardrail``, which returns the inputs untouched on an endpoint
    whose payload it cannot act on) sets ``records_own_guardrail_information =
    True`` so the auto-record is skipped even on the return paths where it
    recorded nothing; otherwise a no-op early return would be logged as an
    "allow"/"success" run even though the guardrail did nothing. The exception
    branch below still records so a genuine failure is not lost.
    """
    import functools
    import inspect

    def _infer_event_type_from_function_name(
        func_name: str,
    ) -> GuardrailEventHooks | None:
        """Infer the actual event type from the function name"""
        if func_name == "async_pre_call_hook":
            return GuardrailEventHooks.pre_call
        elif func_name == "async_moderation_hook":
            return GuardrailEventHooks.during_call
        elif func_name in (
            "async_post_call_success_hook",
            "async_post_call_streaming_hook",
        ):
            return GuardrailEventHooks.post_call
        return None

    @functools.wraps(func)
    async def async_wrapper(*args, **kwargs):
        start_time: Final = datetime.now()  # Move start_time inside the wrapper
        self: Final[CustomGuardrail] = args[0]
        request_data: Final[dict] = kwargs.get("data") or kwargs.get("request_data") or {}
        event_type: Final = _infer_event_type_from_function_name(func.__name__)

        # Store original inputs for comparison (for apply_guardrail functions)
        original_inputs = None
        if func.__name__ == "apply_guardrail" and "inputs" in kwargs:
            original_inputs = kwargs.get("inputs")

        logging_obj: Final = kwargs.get("logging_obj")
        self_recorded_token: Final = _guardrail_self_recorded.set(False)
        try:
            response: Final = await func(*args, **kwargs)
            if self.records_own_guardrail_information or _guardrail_self_recorded.get():
                return response
            return self._process_response(
                response=response,
                request_data=request_data,
                start_time=start_time.timestamp(),
                end_time=datetime.now().timestamp(),
                duration=(datetime.now() - start_time).total_seconds(),
                event_type=event_type,
                original_inputs=original_inputs,
            )
        except Exception as e:
            if _guardrail_self_recorded.get():
                raise
            return self._process_error(
                e=e,
                request_data=request_data,
                start_time=start_time.timestamp(),
                end_time=datetime.now().timestamp(),
                duration=(datetime.now() - start_time).total_seconds(),
                event_type=event_type,
            )
        finally:
            _guardrail_self_recorded.reset(self_recorded_token)
            _sync_guardrail_info_to_logging_obj(request_data, logging_obj)

    @functools.wraps(func)
    def sync_wrapper(*args, **kwargs):
        start_time: Final = datetime.now()  # Move start_time inside the wrapper
        self: Final[CustomGuardrail] = args[0]
        request_data: Final[dict] = kwargs.get("data") or kwargs.get("request_data") or {}
        event_type: Final = _infer_event_type_from_function_name(func.__name__)

        # Store original inputs for comparison (for apply_guardrail functions)
        original_inputs = None
        if func.__name__ == "apply_guardrail" and "inputs" in kwargs:
            original_inputs = kwargs.get("inputs")

        logging_obj: Final = kwargs.get("logging_obj")
        self_recorded_token: Final = _guardrail_self_recorded.set(False)
        try:
            response: Final = func(*args, **kwargs)
            if self.records_own_guardrail_information or _guardrail_self_recorded.get():
                return response
            return self._process_response(
                response=response,
                request_data=request_data,
                duration=(datetime.now() - start_time).total_seconds(),
                event_type=event_type,
                original_inputs=original_inputs,
            )
        except Exception as e:
            if _guardrail_self_recorded.get():
                raise
            return self._process_error(
                e=e,
                request_data=request_data,
                duration=(datetime.now() - start_time).total_seconds(),
                event_type=event_type,
            )
        finally:
            _guardrail_self_recorded.reset(self_recorded_token)
            _sync_guardrail_info_to_logging_obj(request_data, logging_obj)

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if inspect.iscoroutinefunction(func):
            return async_wrapper(*args, **kwargs)
        return sync_wrapper(*args, **kwargs)

    return wrapper
