Merge 6d65ac3e75 into e9b1c11098

litellm/proxy/guardrails/guardrail_hooks/trendai/__init__.py

@@ -0,0 +1,53 @@
+from typing import TYPE_CHECKING, Final
+
+import httpx
+
+import litellm
+from litellm.types.guardrails import GuardrailEventHooks, Mode
+
+from ._models import TrendAISettings
+from .trendai import GUARDRAIL_NAME, TrendAIGuardrail
+
+if TYPE_CHECKING:
+    from litellm.types.guardrails import Guardrail, LitellmParams
+
+
+def _normalize_event_hook(
+    mode: str | list[str] | Mode,  # mutable-ok: guardrail configuration accepts lists of modes
+) -> GuardrailEventHooks | list[GuardrailEventHooks] | Mode:  # mutable-ok: guardrail event hooks require a list
+    if isinstance(mode, str):
+        return GuardrailEventHooks(mode)
+    if isinstance(mode, list):
+        return [GuardrailEventHooks(item) for item in mode]
+    return mode
+
+
+def initialize_guardrail(
+    litellm_params: "LitellmParams", guardrail: "Guardrail", *, async_handler: httpx.AsyncClient | None = None
+) -> TrendAIGuardrail:
+    settings: Final = TrendAISettings.model_validate(litellm_params.model_dump(mode="python", exclude_none=True))
+    guardrail_name: Final = guardrail["guardrail_name"]
+    callback: Final = TrendAIGuardrail(
+        api_key=litellm_params.api_key,
+        api_base=litellm_params.api_base,
+        app_name=settings.app_name,
+        fallback_on_error=settings.fallback_on_error,
+        timeout=settings.timeout,
+        stream_overlap_size=settings.stream_overlap_size,
+        response_content_chunk_size_bytes=settings.response_content_chunk_size_bytes,
+        streaming_buffer_until_moderated=settings.streaming_buffer_until_moderated,
+        async_handler=async_handler,
+        guardrail_name=guardrail_name,
+        event_hook=_normalize_event_hook(litellm_params.mode),
+        default_on=litellm_params.default_on is True,
+    )
+    litellm.logging_callback_manager.add_litellm_callback(  # pyright: ignore[reportUnknownMemberType]  # callback union is partially untyped
+        callback
+    )
+    return callback
+
+
+guardrail_initializer_registry: Final = {GUARDRAIL_NAME: initialize_guardrail}
+guardrail_class_registry: Final = {GUARDRAIL_NAME: TrendAIGuardrail}
+
+__all__ = ("TrendAIGuardrail",)

litellm/proxy/guardrails/guardrail_hooks/trendai/_models.py

@@ -0,0 +1,132 @@
+# Derived from tm-v1-ai-guard-litellm-plugin revision 6cafc143f62962a98d4eb5abe9f608c61ff194d4.
+# This file has been modified for integration into LiteLLM.
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Literal, TypeAlias
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class TrendAISettings(BaseModel, frozen=True):
+    app_name: str | None = None
+    fallback_on_error: Literal["block", "allow"] = "block"
+    timeout: float = 5.0
+    stream_overlap_size: int = 256
+    response_content_chunk_size_bytes: int = 49_500
+    streaming_buffer_until_moderated: bool = True
+
+
+class TrendAIChatMessage(BaseModel, frozen=True):
+    role: Literal["assistant"] = "assistant"
+    content: str
+
+
+class TrendAIChatChoice(BaseModel, frozen=True):
+    index: int = 0
+    message: TrendAIChatMessage
+    finish_reason: Literal["stop"] = "stop"
+
+
+class TrendAIChatCompletionPayload(BaseModel, frozen=True):
+    """The OpenAI chat-completion shape Trend AI Guard scans model output as."""
+
+    id: str = "chatcmpl-stream"
+    object: Literal["chat.completion"] = "chat.completion"
+    created: int = 0
+    model: str
+    choices: tuple[TrendAIChatChoice, ...]
+
+    @classmethod
+    def for_content(cls, content: str, model: str) -> "TrendAIChatCompletionPayload":
+        return cls(model=model, choices=(TrendAIChatChoice(message=TrendAIChatMessage(content=content)),))
+
+
+class TrendAIRedactedMessage(BaseModel, frozen=True):
+    model_config = ConfigDict(extra="ignore")
+
+    content: str | None = None
+
+
+class TrendAIRedactedChoice(BaseModel, frozen=True):
+    model_config = ConfigDict(extra="ignore")
+
+    message: TrendAIRedactedMessage | None = None
+
+
+class TrendAIRedactedResponse(BaseModel, frozen=True):
+    model_config = ConfigDict(extra="ignore")
+
+    choices: tuple[TrendAIRedactedChoice, ...] = ()
+
+
+class TrendAIRedactedPrompt(BaseModel, frozen=True):
+    model_config = ConfigDict(extra="ignore")
+
+    prompt: str | None = None
+
+
+class TrendAISensitiveRule(BaseModel, frozen=True):
+    model_config = ConfigDict(extra="ignore")
+
+    id: str = ""
+
+
+class TrendAISensitiveInformation(BaseModel, frozen=True):
+    model_config = ConfigDict(extra="ignore")
+
+    has_policy_violation: bool = Field(default=False, alias="hasPolicyViolation")
+    rules: tuple[TrendAISensitiveRule, ...] = ()
+
+
+class TrendAIResponse(BaseModel, frozen=True):
+    model_config = ConfigDict(extra="ignore")
+
+    action: str
+    reasons: tuple[str, ...] = ()
+    reason: str = ""
+    redacted_request: Mapping[str, object] | None = Field(default=None, alias="redactedRequest")
+    sensitive_information: TrendAISensitiveInformation | None = Field(default=None, alias="sensitiveInformation")
+
+
+@dataclass(frozen=True, slots=True)
+class TrendAIAllow:
+    kind: Literal["allow"] = "allow"
+    redacted_content: str | None = None
+    masked_entity_count: tuple[tuple[str, int], ...] = ()
+
+
+@dataclass(frozen=True, slots=True)
+class TrendAIBlock:
+    reason: str
+    status_code: int = 400
+    kind: Literal["block"] = "block"
+
+
+@dataclass(frozen=True, slots=True)
+class TrendAIProviderFailure:
+    reason: str
+    status_code: int | None = None
+    kind: Literal["provider_failure"] = "provider_failure"
+
+
+TrendAIScanResult: TypeAlias = TrendAIAllow | TrendAIBlock | TrendAIProviderFailure
+
+
+@dataclass(frozen=True, slots=True)
+class TrendAITextWindow:
+    text: str
+    start: int
+
+
+@dataclass(frozen=True, slots=True)
+class TrendAIWindowScan:
+    window: TrendAITextWindow
+    result: TrendAIScanResult
+
+
+@dataclass(frozen=True, slots=True)
+class TrendAIRequestPrompt:
+    prompt: str
+    texts: tuple[str, ...]
+    message_index: int | None

litellm/proxy/guardrails/guardrail_hooks/trendai/_text.py

@@ -0,0 +1,97 @@
+# Derived from tm-v1-ai-guard-litellm-plugin revision 6cafc143f62962a98d4eb5abe9f608c61ff194d4.
+# This file has been modified for integration into LiteLLM.
+
+from collections.abc import Iterator, Sequence
+from typing import Final
+
+from litellm.llms.base_llm.guardrail_translation.utils import message_slot_texts
+from litellm.types.llms.openai import AllMessageValues
+
+from ._models import TrendAIRequestPrompt, TrendAITextWindow
+
+
+def utf8_windows(content: str, *, chunk_size_bytes: int, overlap_chars: int) -> tuple[TrendAITextWindow, ...]:
+    """Split ``content`` into windows of at most ``chunk_size_bytes`` UTF-8 bytes.
+
+    Consecutive windows share their last ``overlap_chars`` characters so a finding that straddles a
+    window boundary is still seen whole by one scan. The overlap is capped below the window length so
+    every window makes forward progress.
+    """
+    return tuple(_iter_utf8_windows(content, chunk_size_bytes, overlap_chars))
+
+
+def _iter_utf8_windows(content: str, chunk_size_bytes: int, overlap_chars: int) -> Iterator[TrendAITextWindow]:
+    encoded: Final = content.encode("utf-8")
+    byte_offset = 0  # rebind-ok: window cursor advances across the loop
+    char_offset = 0  # rebind-ok: window cursor advances across the loop
+    while byte_offset < len(encoded):
+        text = (
+            encoded[byte_offset : byte_offset + chunk_size_bytes].decode("utf-8", errors="ignore")
+            or content[char_offset]
+        )
+        yield TrendAITextWindow(text=text, start=char_offset)
+        end_byte_offset = byte_offset + len(text.encode("utf-8"))
+        if end_byte_offset >= len(encoded):
+            return
+        overlap = min(max(overlap_chars, 0), len(text) - 1)
+        byte_offset = end_byte_offset - len(text[len(text) - overlap :].encode("utf-8"))
+        char_offset += len(text) - overlap
+
+
+def merge_redaction(original: str, current: str, redacted: str) -> str | None:
+    """Apply the masks ``redacted`` adds over ``original`` on top of the masks already in ``current``.
+
+    Returns None when the three texts do not line up character for character, since a positional
+    merge would then scramble the output.
+    """
+    if len(original) != len(current) or len(redacted) != len(original):
+        return None
+    return "".join(
+        redacted_char if redacted_char != original_char else current_char
+        for original_char, current_char, redacted_char in zip(original, current, redacted, strict=True)
+    )
+
+
+def apply_window_redaction(content: str, window: TrendAITextWindow, redacted: str) -> str | None:
+    end: Final = window.start + len(window.text)
+    merged: Final = merge_redaction(window.text, content[window.start : end], redacted)
+    if merged is None:
+        return None
+    return f"{content[: window.start]}{merged}{content[end:]}"
+
+
+def locate_request_prompt(
+    texts: Sequence[str],
+    structured_messages: Sequence[AllMessageValues] | None,
+) -> TrendAIRequestPrompt | None:
+    """Locate the last user turn and its text slots, or use flat texts when no rows are supplied."""
+    messages: Final = tuple(structured_messages or ())
+    user_index: Final = next(
+        (index for index in reversed(range(len(messages))) if messages[index]["role"] == "user"), None
+    )
+    if messages and user_index is None:
+        return None
+    parts: Final = message_slot_texts(messages[user_index]) if user_index is not None else tuple(texts)
+    prompt: Final = "".join(parts).strip()
+    if not prompt:
+        return None
+    return TrendAIRequestPrompt(prompt=prompt, texts=tuple(parts), message_index=user_index)
+
+
+def redact_request_texts(texts: Sequence[str], redacted: str) -> tuple[str, ...]:
+    """Write a redacted prompt back over the ``texts`` entries the prompt was assembled from.
+
+    A single-part prompt is replaced outright. A multi-part prompt whose redaction kept its length
+    is split back into the original parts; otherwise the first part carries the whole redaction and
+    the rest are blanked, so no original part can leak.
+    """
+    parts: Final = tuple(texts)
+    if len(parts) == 1:
+        return (redacted,)
+    joined: Final = "".join(parts)
+    leading: Final = len(joined) - len(joined.lstrip())
+    aligned: Final = f"{joined[:leading]}{redacted}{joined[len(joined.rstrip()) :]}"
+    if len(aligned) != len(joined):
+        return (redacted, *("" for _ in parts[1:]))
+    offsets: Final = tuple(sum(len(part) for part in parts[:index]) for index in range(len(parts) + 1))
+    return tuple(aligned[offsets[index] : offsets[index + 1]] for index in range(len(parts)))

litellm/proxy/guardrails/guardrail_hooks/trendai/trendai.py

@@ -0,0 +1,520 @@
+# Derived from tm-v1-ai-guard-litellm-plugin revision 6cafc143f62962a98d4eb5abe9f608c61ff194d4.
+# This file has been modified for integration into LiteLLM.
+
+import asyncio
+import os
+import time
+from collections.abc import AsyncIterator, Mapping, Sequence
+from itertools import chain
+from types import MappingProxyType
+from typing import TYPE_CHECKING, ClassVar, Final, Literal, NoReturn, Protocol
+from urllib.parse import SplitResult, urlsplit, urlunsplit
+
+import httpx
+from pydantic import ValidationError
+from typing_extensions import assert_never
+
+from litellm._logging import verbose_proxy_logger
+from litellm._version import version as litellm_version
+from litellm.exceptions import GuardrailRaisedException, Timeout
+from litellm.integrations.custom_guardrail import (
+    CustomGuardrail,
+    log_guardrail_information,  # pyright: ignore[reportUnknownVariableType]  # legacy decorator has an untyped signature
+)
+from litellm.llms.base_llm.guardrail_translation.utils import message_with_slot_texts
+from litellm.llms.custom_httpx.http_handler import (
+    get_async_httpx_client,  # pyright: ignore[reportUnknownVariableType]  # legacy client factory has an untyped params map
+    httpxSpecialProvider,
+)
+from litellm.secret_managers.main import get_secret_str
+from litellm.types.guardrails import GuardrailEventHooks, Mode
+from litellm.types.utils import GenericGuardrailAPIInputs, GuardrailStatus
+
+from ._models import (
+    TrendAIAllow,
+    TrendAIBlock,
+    TrendAIChatCompletionPayload,
+    TrendAIProviderFailure,
+    TrendAIRedactedPrompt,
+    TrendAIRedactedResponse,
+    TrendAIResponse,
+    TrendAIScanResult,
+    TrendAIWindowScan,
+)
+from ._text import apply_window_redaction, locate_request_prompt, redact_request_texts, utf8_windows
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+
+GUARDRAIL_NAME: Final = "trendai"
+OPENAI_CHAT_COMPLETION_RESPONSE_V1: Final = "OpenAIChatCompletionResponseV1"
+RESPONSE_CONTENT_CHUNK_SIZE_BYTES: Final = 49_500
+TMV1_CLIENT_NAME: Final = "litellm"
+PLUGIN_VERSION: Final = "v0.1.3"
+PROVIDER_UNAVAILABLE_STATUS: Final = 503
+_APPLY_GUARDRAILS_PATH: Final = "/applyGuardrails"
+_TREND_AI_SECURITY_PATH: Final = "/v3.0/aiSecurity"
+_RESPONSE_MODEL: Final = "guardrailed-response"
+_REDACTION_FAILED_MESSAGE: Final = "Trend AI Guard could not apply the requested redaction"
+_NO_ENTITIES: Final[Mapping[str, int]] = MappingProxyType({})
+
+
+class _AsyncHTTPClient(Protocol):
+    async def post(
+        self,
+        url: str,
+        *,
+        json: dict[str, object] | None = None,  # mutable-ok: injected HTTP handler requires dict payloads
+        headers: dict[str, str] | None = None,  # mutable-ok: injected HTTP handler requires dict headers
+        timeout: float | httpx.Timeout | None = None,
+    ) -> httpx.Response: ...
+
+
+class TrendAIGuardrail(CustomGuardrail):
+    records_own_guardrail_information: ClassVar[bool] = True
+
+    def __init__(
+        self,
+        api_key: str | None = None,
+        api_base: str | None = None,
+        app_name: str | None = None,
+        fallback_on_error: Literal["block", "allow"] = "block",
+        timeout: float = 5.0,
+        stream_overlap_size: int = 256,
+        response_content_chunk_size_bytes: int = RESPONSE_CONTENT_CHUNK_SIZE_BYTES,
+        async_handler: _AsyncHTTPClient | None = None,
+        guardrail_name: str | None = None,
+        event_hook: (  # mutable-ok: CustomGuardrail accepts lists of hooks
+            GuardrailEventHooks | list[GuardrailEventHooks] | Mode | None
+        ) = None,
+        default_on: bool = False,
+        streaming_buffer_until_moderated: bool = True,
+    ) -> None:
+        resolved_api_key: Final = api_key or get_secret_str("TMV1_API_KEY")
+        if not resolved_api_key:
+            raise ValueError(
+                "Trend AI Guard requires an API key. Pass api_key or set the TMV1_API_KEY environment variable."
+            )
+
+        resolved_api_base: Final = api_base or os.environ.get("TRENDAI_AI_GUARD_BASE_URL")
+        if not resolved_api_base:
+            raise ValueError(
+                "Trend AI Guard requires an API base URL. Pass api_base or set the "
+                "TRENDAI_AI_GUARD_BASE_URL environment variable."
+            )
+        if fallback_on_error not in ("block", "allow"):
+            raise ValueError("fallback_on_error must be 'block' or 'allow'")
+        if timeout <= 0:
+            raise ValueError("timeout must be greater than zero")
+        if stream_overlap_size < 0:
+            raise ValueError("stream_overlap_size must be non-negative")
+        if response_content_chunk_size_bytes < 1:
+            raise ValueError("response_content_chunk_size_bytes must be greater than zero")
+
+        self.api_key: str = resolved_api_key
+        self.api_url: str = _build_apply_guardrails_url(resolved_api_base)
+        self.app_name: str = app_name or os.environ.get("TMV1_APPLICATION_NAME", "litellm")
+        self.fallback_on_error: Literal["block", "allow"] = fallback_on_error
+        self.timeout = timeout
+        self.stream_overlap_size: int = stream_overlap_size
+        self.response_content_chunk_size_bytes: int = response_content_chunk_size_bytes
+        self.streaming_buffer_until_moderated: bool = streaming_buffer_until_moderated
+        self.async_handler: _AsyncHTTPClient = async_handler or get_async_httpx_client(
+            llm_provider=httpxSpecialProvider.GuardrailCallback
+        )
+
+        super().__init__(  # pyright: ignore[reportUnknownMemberType]  # base constructor retains untyped extension kwargs
+            guardrail_name=guardrail_name,
+            event_hook=event_hook,
+            default_on=default_on,
+            supported_event_hooks=self.get_supported_event_hooks(),
+        )
+
+    @classmethod
+    def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:  # mutable-ok: CustomGuardrail contract
+        return [
+            GuardrailEventHooks.pre_call,
+            GuardrailEventHooks.during_call,
+            GuardrailEventHooks.post_call,
+            GuardrailEventHooks.logging_only,
+        ]
+
+    def supports_scan_only_tool_results(self) -> bool:
+        return False
+
+    @log_guardrail_information
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict[str, object],  # mutable-ok: CustomGuardrail hook contract is a plain dict
+        input_type: Literal["request", "response"],
+        logging_obj: "LiteLLMLoggingObj | None" = None,
+    ) -> GenericGuardrailAPIInputs:
+        texts: Final = tuple(inputs.get("texts") or ())
+        match input_type:
+            case "request":
+                return await self._apply_request_guardrail(inputs, texts, request_data)
+            case "response":
+                return await self._apply_response_guardrail(inputs, texts, request_data)
+            case _:
+                assert_never(input_type)
+
+    async def _apply_request_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        texts: tuple[str, ...],
+        request_data: dict[str, object],  # mutable-ok: CustomGuardrail hook contract is a plain dict
+    ) -> GenericGuardrailAPIInputs:
+        messages: Final = tuple(inputs.get("structured_messages") or ())
+        prompt: Final = locate_request_prompt(texts, messages)
+        if prompt is None:
+            verbose_proxy_logger.debug("Trend AI Guard: no user prompt to scan in request inputs")
+            return inputs
+        started_at: Final = time.time()
+        result: Final = await self._scan_payload(MappingProxyType({"prompt": prompt.prompt}))
+        self._record_scan(result, request_data, started_at, event_type=None)
+        redacted: Final = self._enforce(result)
+        if redacted is None:
+            return inputs
+        redacted_texts: Final = redact_request_texts(prompt.texts, redacted)
+        if prompt.message_index is None:
+            return {**inputs, "texts": list(redacted_texts)}
+        rewritten_message: Final = message_with_slot_texts(messages[prompt.message_index], redacted_texts)
+        if rewritten_message is None:
+            self._raise_redaction_failure(request_data, started_at, event_type=None)
+        return {
+            **inputs,
+            "structured_messages": [
+                *messages[: prompt.message_index],
+                rewritten_message,
+                *messages[prompt.message_index + 1 :],
+            ],
+        }
+
+    async def _apply_response_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        texts: tuple[str, ...],
+        request_data: dict[str, object],  # mutable-ok: CustomGuardrail hook contract is a plain dict
+    ) -> GenericGuardrailAPIInputs:
+        if not texts:
+            verbose_proxy_logger.debug("Trend AI Guard: no response text to scan")
+            return inputs
+        model: Final = inputs.get("model") or _RESPONSE_MODEL
+        started_at: Final = time.time()
+        scans: Final = tuple([await self._scan_response_windows(text, model) for text in texts])
+        all_scans: Final = tuple(chain.from_iterable(scans))
+        verdict: Final = _window_verdict(all_scans)
+        self._record_scan(
+            verdict,
+            request_data,
+            started_at,
+            event_type=None,
+            redacted=any(_is_redacting(scan.result) for scan in all_scans),
+        )
+        self._enforce(verdict)
+        redacted_texts: Final = tuple(
+            _merge_window_redactions(text, text_scans) for text, text_scans in zip(texts, scans, strict=True)
+        )
+        merged_texts: Final = tuple(text for text in redacted_texts if text is not None)
+        if len(merged_texts) != len(texts):
+            self._raise_redaction_failure(request_data, started_at, event_type=None)
+        redacted_inputs: Final[GenericGuardrailAPIInputs] = {
+            **inputs,
+            "texts": list(merged_texts),
+        }
+        return redacted_inputs
+
+    async def _scan_response_windows(self, content: str, model: str) -> tuple[TrendAIWindowScan, ...]:
+        return tuple([scan async for scan in self._iter_response_window_scans(content, model)])
+
+    async def _iter_response_window_scans(self, content: str, model: str) -> AsyncIterator[TrendAIWindowScan]:
+        """Scan ``content`` window by window, stopping at the first block or provider failure."""
+        for window in utf8_windows(
+            content,
+            chunk_size_bytes=self.response_content_chunk_size_bytes,
+            overlap_chars=self.stream_overlap_size,
+        ):
+            result = await self._scan_payload(
+                _chat_completion_payload(window.text, model),
+                request_type=OPENAI_CHAT_COMPLETION_RESPONSE_V1,
+            )
+            yield TrendAIWindowScan(window=window, result=result)
+            if not isinstance(result, TrendAIAllow):
+                return
+
+    def _raise_redaction_failure(
+        self,
+        request_data: dict[str, object],  # mutable-ok: CustomGuardrail hook contract is a plain dict
+        started_at: float,
+        *,
+        event_type: GuardrailEventHooks | None,
+    ) -> NoReturn:
+        self._record_failure(_REDACTION_FAILED_MESSAGE, request_data, started_at, event_type=event_type)
+        raise GuardrailRaisedException(
+            guardrail_name=self.guardrail_name,
+            message=_REDACTION_FAILED_MESSAGE,
+            should_wrap_with_default_message=False,
+            status_code=PROVIDER_UNAVAILABLE_STATUS,
+        )
+
+    def _enforce(self, result: TrendAIScanResult) -> str | None:
+        """Raise for a block or a fail-closed provider failure; otherwise return any redacted content."""
+        match result:
+            case TrendAIAllow(redacted_content=redacted_content):
+                return redacted_content
+            case TrendAIBlock(reason=reason, status_code=status_code):
+                raise GuardrailRaisedException(
+                    guardrail_name=self.guardrail_name,
+                    message=f"Blocked by Trend AI Guard. Security violation: {reason}",
+                    should_wrap_with_default_message=False,
+                    status_code=status_code,
+                    blocked_content=True,
+                )
+            case TrendAIProviderFailure(reason=reason, status_code=status_code):
+                if self.fallback_on_error == "allow":
+                    verbose_proxy_logger.warning(
+                        "Trend AI Guard: %s (status=%s); allowing traffic (fallback_on_error=allow)",
+                        reason,
+                        status_code,
+                    )
+                    return None
+                verbose_proxy_logger.error(
+                    "Trend AI Guard: %s (status=%s); blocking traffic (fallback_on_error=block)", reason, status_code
+                )
+                raise GuardrailRaisedException(
+                    guardrail_name=self.guardrail_name,
+                    message=f"Security Guard Error: {reason}",
+                    should_wrap_with_default_message=False,
+                    status_code=PROVIDER_UNAVAILABLE_STATUS,
+                )
+            case _:
+                assert_never(result)
+
+    def _record_scan(
+        self,
+        result: TrendAIScanResult,
+        request_data: dict[str, object],  # mutable-ok: CustomGuardrail hook contract is a plain dict
+        started_at: float,
+        *,
+        event_type: GuardrailEventHooks | None,
+        redacted: bool | None = None,
+    ) -> None:
+        match result:
+            case TrendAIAllow(redacted_content=redacted_content, masked_entity_count=masked_entity_count):
+                self._record(
+                    _allow_record(redacted_content is not None if redacted is None else redacted),
+                    "success",
+                    request_data,
+                    started_at,
+                    event_type=event_type,
+                    masked_entity_count=_merge_entity_counts(_NO_ENTITIES, masked_entity_count),
+                )
+            case TrendAIBlock(reason=reason):
+                self._record(
+                    MappingProxyType({"action": "block", "reason": reason}),
+                    "guardrail_intervened",
+                    request_data,
+                    started_at,
+                    event_type=event_type,
+                )
+            case TrendAIProviderFailure(reason=reason, status_code=status_code):
+                self._record(
+                    MappingProxyType(
+                        {"description": reason, "status_code": status_code, "fallback_on_error": self.fallback_on_error}
+                    ),
+                    "guardrail_failed_to_respond",
+                    request_data,
+                    started_at,
+                    event_type=event_type,
+                )
+            case _:
+                assert_never(result)
+
+    def _record_failure(
+        self,
+        description: str,
+        request_data: dict[str, object],  # mutable-ok: CustomGuardrail hook contract is a plain dict
+        started_at: float,
+        *,
+        event_type: GuardrailEventHooks | None,
+    ) -> None:
+        self._record(
+            MappingProxyType({"description": description}),
+            "guardrail_failed_to_respond",
+            request_data,
+            started_at,
+            event_type=event_type,
+        )
+
+    def _record(
+        self,
+        guardrail_json_response: Mapping[str, object],
+        guardrail_status: GuardrailStatus,
+        request_data: dict[str, object],  # mutable-ok: CustomGuardrail hook contract is a plain dict
+        started_at: float,
+        *,
+        event_type: GuardrailEventHooks | None,
+        masked_entity_count: Mapping[str, int] = _NO_ENTITIES,
+    ) -> None:
+        ended_at: Final = time.time()
+        self.add_standard_logging_guardrail_information_to_request_data(  # pyright: ignore[reportUnknownMemberType]  # base signature takes an untyped dict
+            guardrail_json_response=dict(guardrail_json_response),
+            request_data=request_data,
+            guardrail_status=guardrail_status,
+            start_time=started_at,
+            end_time=ended_at,
+            duration=ended_at - started_at,
+            event_type=event_type,
+            masked_entity_count=dict(masked_entity_count) or None,
+        )
+
+    def _build_request_headers(self, request_type: str | None = None) -> Mapping[str, str]:
+        headers: Final = (
+            ("TMV1-Application-Name", self.app_name),
+            ("Authorization", f"Bearer {self.api_key}"),
+            ("Content-Type", "application/json"),
+            ("TMV1-Client-Name", TMV1_CLIENT_NAME),
+            ("TMV1-Client-Version", litellm_version),
+            ("TMV1-Plugin-Version", PLUGIN_VERSION),
+            ("prefer", "redact-pii,return=representation"),
+            *_optional_header("TMV1-Request-Type", request_type),
+        )
+        return MappingProxyType(dict(headers))
+
+    async def _scan_payload(
+        self,
+        payload: Mapping[str, object],
+        request_type: str | None = None,
+    ) -> TrendAIScanResult:
+        try:
+            response: Final = await self.async_handler.post(
+                self.api_url,
+                json=dict(payload),
+                headers=dict(self._build_request_headers(request_type)),
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            parsed: Final = TrendAIResponse.model_validate_json(response.content)
+        except httpx.HTTPStatusError as error:
+            return TrendAIProviderFailure(
+                reason="Trend AI Guard returned an HTTP error",
+                status_code=error.response.status_code,
+            )
+        except (httpx.RequestError, Timeout, asyncio.TimeoutError) as error:
+            return TrendAIProviderFailure(reason=f"Trend AI Guard request failed: {type(error).__name__}")
+        except ValidationError:
+            return TrendAIProviderFailure(reason="Trend AI Guard returned an invalid response")
+
+        normalized_action: Final = parsed.action.strip().lower()
+        if normalized_action == "block":
+            reason: Final = ", ".join(parsed.reasons) or parsed.reason or "Content policy violation"
+            return TrendAIBlock(reason=reason)
+        if normalized_action != "allow":
+            return TrendAIProviderFailure(reason=f"Trend AI Guard returned an unsupported action: {parsed.action!r}")
+
+        redacted_content: Final = _extract_redacted_content(request_type, parsed)
+        if parsed.redacted_request is not None and redacted_content is None:
+            return TrendAIProviderFailure(reason="Trend AI Guard returned an invalid redacted payload")
+
+        entity_counts: Final = (
+            tuple(
+                sorted(
+                    (
+                        rule_id,
+                        sum(1 for candidate in parsed.sensitive_information.rules if candidate.id.strip() == rule_id),
+                    )
+                    for rule_id in frozenset(
+                        rule.id.strip() for rule in parsed.sensitive_information.rules if rule.id.strip()
+                    )
+                )
+            )
+            if parsed.sensitive_information is not None and redacted_content is not None
+            else ()
+        )
+        return TrendAIAllow(redacted_content=redacted_content, masked_entity_count=entity_counts)
+
+
+def _chat_completion_payload(content: str, model: str) -> Mapping[str, object]:
+    return MappingProxyType(TrendAIChatCompletionPayload.for_content(content, model).model_dump())
+
+
+def _optional_header(name: str, value: str | None) -> tuple[tuple[str, str], ...]:
+    return () if value is None else ((name, value),)
+
+
+def _allow_record(redacted: bool) -> Mapping[str, object]:
+    return MappingProxyType({"action": "allow", "redacted": redacted})
+
+
+def _is_redacting(result: TrendAIScanResult) -> bool:
+    return isinstance(result, TrendAIAllow) and result.redacted_content is not None
+
+
+def _window_verdict(scans: Sequence[TrendAIWindowScan]) -> TrendAIScanResult:
+    """Collapse window scans into one verdict: the first non-allow result, else an allow carrying every entity count."""
+    terminal: Final = next((scan.result for scan in scans if not isinstance(scan.result, TrendAIAllow)), None)
+    if terminal is not None:
+        return terminal
+    allowed_counts: Final = (scan.result.masked_entity_count for scan in scans if isinstance(scan.result, TrendAIAllow))
+    counts: Final = tuple(chain.from_iterable(allowed_counts))
+    return TrendAIAllow(masked_entity_count=tuple(sorted(_merge_entity_counts(_NO_ENTITIES, counts).items())))
+
+
+def _merge_entity_counts(counts: Mapping[str, int], additions: Sequence[tuple[str, int]]) -> Mapping[str, int]:
+    """Per-entity maxima: overlapping scans may report the same entity more than once."""
+    entities: Final = frozenset(counts) | frozenset(entity for entity, _ in additions)
+    return MappingProxyType(
+        {
+            entity: max((counts.get(entity, 0), *(count for candidate, count in additions if candidate == entity)))
+            for entity in entities
+        }
+    )
+
+
+def _merge_window_redactions(content: str, scans: Sequence[TrendAIWindowScan]) -> str | None:
+    """Fold every window's redaction into ``content``; None if any window cannot be merged positionally."""
+    merged = content  # rebind-ok: folded across the windows
+    for scan in scans:
+        if not isinstance(scan.result, TrendAIAllow) or scan.result.redacted_content is None:
+            continue
+        merged = apply_window_redaction(merged, scan.window, scan.result.redacted_content)
+        if merged is None:
+            return None
+    return merged
+
+
+def _build_apply_guardrails_url(api_base: str) -> str:
+    parsed: Final = urlsplit(api_base.strip())
+    normalized_path: Final = parsed.path.rstrip("/")
+    path_with_product: Final = (
+        f"{normalized_path}{_TREND_AI_SECURITY_PATH}"
+        if parsed.hostname is not None
+        and parsed.hostname.lower().endswith(".trendmicro.com")
+        and "/aiSecurity" not in normalized_path
+        else normalized_path
+    )
+    final_path: Final = (
+        path_with_product
+        if path_with_product.endswith(_APPLY_GUARDRAILS_PATH)
+        else f"{path_with_product}{_APPLY_GUARDRAILS_PATH}"
+    )
+    return urlunsplit(SplitResult(parsed.scheme, parsed.netloc, final_path, parsed.query, parsed.fragment))
+
+
+def _extract_redacted_content(request_type: str | None, response: TrendAIResponse) -> str | None:
+    payload: Final = response.redacted_request
+    if payload is None:
+        return None
+    try:
+        if request_type == OPENAI_CHAT_COMPLETION_RESPONSE_V1:
+            parsed_response: Final = TrendAIRedactedResponse.model_validate(payload)
+            content: Final = " ".join(
+                choice.message.content
+                for choice in parsed_response.choices
+                if choice.message is not None and choice.message.content
+            )
+            return content or None
+        return TrendAIRedactedPrompt.model_validate(payload).prompt or None
+    except ValidationError:
+        return None