diff --git a/litellm/proxy/guardrails/guardrail_hooks/privaite/__init__.py b/litellm/proxy/guardrails/guardrail_hooks/privaite/__init__.py
new file mode 100644
index 00000000000..630c3c2e1d2
--- /dev/null
+++ b/litellm/proxy/guardrails/guardrail_hooks/privaite/__init__.py
@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Final
+
+from litellm.types.guardrails import SupportedGuardrailIntegrations
+
+from .privaite import PrivaiteGuardrail
+
+if TYPE_CHECKING:
+    from litellm.types.guardrails import Guardrail, LitellmParams
+
+
+def initialize_guardrail(litellm_params: LitellmParams, guardrail: Guardrail) -> PrivaiteGuardrail:
+    import litellm
+
+    _privaite_callback: Final = PrivaiteGuardrail(
+        guardrail_name=guardrail["guardrail_name"],
+        event_hook=litellm_params.mode,
+        default_on=litellm_params.default_on,
+        preset=litellm_params.preset,
+        languages=litellm_params.languages,
+        deanonymize=litellm_params.deanonymize,
+        block_entities=litellm_params.block_entities,
+    )
+    litellm.logging_callback_manager.add_litellm_callback(_privaite_callback)
+    return _privaite_callback
+
+
+guardrail_initializer_registry: Final = {
+    SupportedGuardrailIntegrations.PRIVAITE.value: initialize_guardrail,
+}
+
+guardrail_class_registry: Final = {
+    SupportedGuardrailIntegrations.PRIVAITE.value: PrivaiteGuardrail,
+}
diff --git a/litellm/proxy/guardrails/guardrail_hooks/privaite/engine.py b/litellm/proxy/guardrails/guardrail_hooks/privaite/engine.py
new file mode 100644
index 00000000000..3cc873f7b91
--- /dev/null
+++ b/litellm/proxy/guardrails/guardrail_hooks/privaite/engine.py
@@ -0,0 +1,159 @@
+"""The typed boundary to the optional `privaite` package.
+
+`privaite` is not a LiteLLM dependency, so it is imported lazily and nothing it returns
+has a type here. Each object it hands back is checked once against the protocol below
+that names what this guardrail calls on it.
+"""
+
+import asyncio
+import importlib
+from collections.abc import Callable, Iterable, Mapping, Sequence
+from types import MappingProxyType, ModuleType
+from typing import Final, Protocol, TypeVar, runtime_checkable
+
+INSTALL_HINT: Final = (
+    "The PrivAiTe guardrail requires the 'privaite' package. Install it with: pip install 'privaite>=0.7.2'"
+)
+
+SPACY_MODELS: Final = MappingProxyType(
+    {
+        "en": "en_core_web_lg",
+        "fr": "fr_core_news_md",
+        "de": "de_core_news_md",
+        "es": "es_core_news_md",
+        "it": "it_core_news_md",
+        "pt": "pt_core_news_md",
+        "nl": "nl_core_news_md",
+    }
+)
+
+_Narrowed = TypeVar("_Narrowed")  # rebind-ok: a TypeVar cannot be declared Final
+
+
+@runtime_checkable
+class RestoreMapping(Protocol):
+    @property
+    def is_empty(self) -> bool: ...
+
+    def add(self, original: str, fake: str, entity_type: str) -> None: ...
+
+    def get_all_fakes(self) -> Mapping[str, str]: ...
+
+
+@runtime_checkable
+class PrivaiteEngine(Protocol):
+    async def initialize(self) -> None: ...
+
+    async def process_request(self, messages: Sequence[object]) -> tuple[Sequence[object], RestoreMapping]: ...
+
+    async def process_response(self, text: str, mapping: RestoreMapping) -> str: ...
+
+
+@runtime_checkable
+class RestoreBuffer(Protocol):
+    def feed(self, text: str) -> str: ...
+
+    def flush(self) -> str: ...
+
+
+def _module(name: str) -> ModuleType:
+    try:
+        return importlib.import_module(name)
+    except ImportError as exc:
+        raise RuntimeError(INSTALL_HINT) from exc
+
+
+def _factory(module: ModuleType, name: str) -> Callable[..., object]:
+    value: Final = getattr(module, name, None)
+    if callable(value):
+        return value
+    if hasattr(module, name):
+        raise TypeError(f"{module.__name__}.{name} is not callable. {INSTALL_HINT}")
+    raise RuntimeError(f"{module.__name__}.{name} is missing. {INSTALL_HINT}")
+
+
+def _narrow(value: object, kind: type[_Narrowed]) -> _Narrowed:
+    if not isinstance(value, kind):
+        raise TypeError(f"privaite returned {type(value).__name__} where {kind.__name__} was expected")
+    return value
+
+
+def _declares_field(model: object, name: str) -> bool:
+    fields: Final = getattr(model, "model_fields", None)
+    return isinstance(fields, Mapping) and name in fields
+
+
+def _is_missing_spacy_model(exc: OSError) -> bool:
+    """spaCy's E050. Permission, disk and network errors must not trigger a download."""
+    message: Final = str(exc)
+    return "E050" in message or "Can't find model" in message
+
+
+async def _download_spacy_models(languages: Sequence[str]) -> None:
+    """Pulls hundreds of MB through pip's synchronous machinery, so it runs off the event loop."""
+    download: Final = _factory(importlib.import_module("spacy.cli"), "download")
+    for language in languages:
+        model = SPACY_MODELS.get(language)
+        if model:
+            await asyncio.to_thread(download, model)
+
+
+async def build_engine(
+    preset: str, languages: Sequence[str], deanonymize: bool, block_entities: Sequence[str]
+) -> PrivaiteEngine:
+    schema: Final = _module("privaite.config.schema")
+    make_engine: Final = _factory(_module("privaite.pii.engine"), "PIIEngine")
+    pii_config: Final = _factory(schema, "PIIConfig")
+    if block_entities and not _declares_field(pii_config, "block_entities"):
+        # PIIConfig allows extra fields, so an older privaite would accept block_entities,
+        # ignore it and forward the PII. Refusing is the only safe answer.
+        raise RuntimeError(
+            "block_entities is set but the installed privaite does not support it; "
+            "upgrade privaite to a version that enforces pii.block_entities."
+        )
+    presidio: Final = _factory(schema, "PresidioDetectorConfig")(enabled=True, languages=list(languages))
+    config: Final = pii_config(
+        enabled=True,
+        preset=preset,
+        detectors=_factory(schema, "DetectorsConfig")(presidio=presidio),
+        anonymization=_factory(schema, "AnonymizationConfig")(method="placeholder"),
+        deanonymization=_factory(schema, "DeanonymizationConfig")(enabled=deanonymize),
+        block_entities=list(block_entities),
+    )
+    engine: Final = _narrow(make_engine(config), PrivaiteEngine)
+    try:
+        await engine.initialize()
+    except OSError as exc:
+        if not _is_missing_spacy_model(exc):
+            raise
+        await _download_spacy_models(languages)
+        retried: Final = _narrow(make_engine(config), PrivaiteEngine)
+        await retried.initialize()
+        return retried
+    return engine
+
+
+def blocked_error() -> type[Exception]:
+    """The exception privaite raises when a request carries a type listed in block_entities."""
+    error: Final = getattr(_module("privaite.pii.engine"), "PIIBlockedError", None)
+    if not isinstance(error, type) or not issubclass(error, Exception):
+        raise TypeError(f"privaite.pii.engine.PIIBlockedError is missing. {INSTALL_HINT}")
+    return error
+
+
+def restore_mapping(pairs: Iterable[tuple[str, str]]) -> RestoreMapping:
+    """A privaite mapping holding each (fake, original) pair."""
+    mapping: Final = _narrow(_factory(_module("privaite.pii.mapping"), "PIIMapping")(), RestoreMapping)
+    for fake, original in pairs:
+        mapping.add(original, fake, "PII")
+    return mapping
+
+
+def restore_buffer_factory() -> Callable[[RestoreMapping], RestoreBuffer]:
+    """privaite's streaming de-anonymizer, which holds back a placeholder split across chunks."""
+    make_buffer: Final = _factory(_module("privaite.streaming.buffer"), "StreamingDeAnonymizer")
+
+    def _buffer(mapping: RestoreMapping) -> RestoreBuffer:
+        return _narrow(make_buffer(mapping), RestoreBuffer)
+
+    return _buffer
diff --git a/litellm/proxy/guardrails/guardrail_hooks/privaite/payload.py b/litellm/proxy/guardrails/guardrail_hooks/privaite/payload.py
new file mode 100644
index 00000000000..e3b6df66a3c
--- /dev/null
+++ b/litellm/proxy/guardrails/guardrail_hooks/privaite/payload.py
@@ -0,0 +1,61 @@
+import functools
+from collections.abc import Callable, Sequence
+from typing import Final, TypeAlias
+
+JsonObject: TypeAlias = dict[str, object]
+
+JsonArray: TypeAlias = list[object]
+
+Write: TypeAlias = Callable[[str], None]
+
+Slot: TypeAlias = tuple[str, Write]
+
+
+def as_object(value: object) -> JsonObject | None:
+    """`value` as a JSON object, or None. A JSON object's keys are strings, which isinstance cannot state."""
+    return value if isinstance(value, dict) else None  # pyright: ignore[reportUnknownVariableType]  # JSON keys are str
+
+
+def as_array(value: object) -> JsonArray | None:
+    """`value` as a JSON array, or None. See `as_object`."""
+    return value if isinstance(value, list) else None  # pyright: ignore[reportUnknownVariableType]  # see as_object
+
+
+def read_field(holder: object, name: str) -> object:
+    """Reads one field from a dict or from an object.
+
+    LiteLLM hands the hooks Pydantic models on some paths and plain dicts on others,
+    so every walk here reads both shapes.
+    """
+    fields: Final = as_object(holder)
+    if fields is not None:
+        return fields.get(name)
+    return getattr(holder, name, None)
+
+
+def read_list(holder: object, name: str) -> Sequence[object]:
+    """Reads a list field from a dict or an object. Anything else reads as empty."""
+    return as_array(read_field(holder, name)) or ()
+
+
+def write_field(holder: object, name: str, value: object) -> None:
+    """Writes one field back into a dict or an object. Pairs with read_field."""
+    fields: Final = as_object(holder)
+    if fields is not None:
+        fields[name] = value
+    else:
+        setattr(holder, name, value)
+
+
+def field_writer(fields: JsonObject, key: str) -> Write:
+    return functools.partial(fields.__setitem__, key)
+
+
+def item_writer(items: JsonArray, index: int) -> Write:
+    return functools.partial(items.__setitem__, index)
+
+
+def index_of(holder: object) -> int:
+    """A streamed choice or tool call is matched across chunks by its index, 0 when absent."""
+    index: Final = read_field(holder, "index")
+    return index if isinstance(index, int) else 0
diff --git a/litellm/proxy/guardrails/guardrail_hooks/privaite/privaite.py b/litellm/proxy/guardrails/guardrail_hooks/privaite/privaite.py
new file mode 100644
index 00000000000..4ec7c3f0748
--- /dev/null
+++ b/litellm/proxy/guardrails/guardrail_hooks/privaite/privaite.py
@@ -0,0 +1,293 @@
+"""
+PrivAiTe guardrail for the LiteLLM proxy.
+
+Runs PrivAiTe's engine in-process. The pre-call hook anonymizes the request and stashes
+the reversible map in the request metadata, and the post-call hooks consume it to restore
+the real values in the response, streaming included. The request surface scanned matches
+the PrivAiTe core: chat `messages`, every Responses `input` item shape and
+`prompt.variables`, the `/v1/completions` `prompt` and `suffix`, and the chat
+`prediction.content` and `web_search_options.user_location` fields. Responses API
+streaming restore is not implemented yet.
+
+If `block_entities` is configured, a request containing any of those PII types is
+rejected with a 400 before anything is forwarded. The error names the types, never the
+values.
+
+There is no detection or masking logic here, it all lives in the `privaite` package,
+which is imported lazily so this module stays importable for the guardrail registry
+without it. See https://github.com/crp4222/PrivAiTe.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import itertools
+from collections.abc import AsyncGenerator, AsyncIterable, Iterable, Iterator, Sequence
+from typing import TYPE_CHECKING, Final, TypeAlias
+
+from typing_extensions import TypedDict, Unpack
+
+from litellm.integrations.custom_guardrail import CustomGuardrail
+from litellm.types.guardrails import GuardrailEventHooks, Mode
+
+from .engine import (
+    PrivaiteEngine,
+    RestoreMapping,
+    blocked_error,
+    build_engine,
+    restore_buffer_factory,
+    restore_mapping,
+)
+from .payload import JsonObject, as_array, as_object, read_field, read_list, write_field
+from .request_walk import (
+    aux_field_slots,
+    has_aux_fields,
+    prompt_variable_slots,
+    prompt_variables,
+    responses_input_slots,
+)
+from .restore import (
+    StreamRestorer,
+    json_escape,
+    restore_message,
+    restore_responses_output,
+)
+
+if TYPE_CHECKING:
+    from litellm.caching.caching import DualCache
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.types.utils import ModelResponseStream
+
+RESTORE_MAP_KEY: Final = "privaite_map"
+
+_PRESETS: Final = frozenset({"light", "onnx"})
+
+_FALSE_STRINGS: Final = frozenset({"false", "0", "no", ""})
+
+_REQUIRED_HOOKS: Final = (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call)
+
+_EngineKey: TypeAlias = tuple[str, tuple[str, ...], bool, tuple[str, ...]]
+
+
+class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
+    pass
+
+
+def _parse_block_entities(raw: str | Sequence[str] | None) -> tuple[str, ...]:
+    entries: Final = raw.split(",") if isinstance(raw, str) else tuple(str(entry) for entry in raw or ())
+    return tuple(entry.strip() for entry in entries if entry.strip())
+
+
+def _with_required_hooks(
+    hook: GuardrailEventHooks | Sequence[GuardrailEventHooks] | None,
+) -> list[GuardrailEventHooks]:  # mutable-ok: CustomGuardrail.event_hook is a list
+    """Both hooks are mandatory. A `mode: post_call` config alone would skip the
+    anonymization and forward raw PII to the model."""
+    configured: Final = ((hook,) if hook else ()) if isinstance(hook, str) else tuple(hook or ())
+    return list(dict.fromkeys((*configured, *_REQUIRED_HOOKS)))
+
+
+def _drop_restore_map(request: object) -> None:
+    metadata: Final = as_object(read_field(request, "metadata"))
+    if metadata is not None:
+        metadata.pop(RESTORE_MAP_KEY, None)
+
+
+def _pop_restore_map(request: JsonObject) -> tuple[tuple[str, str], ...]:
+    """Consumes the reversible map, so the originals do not linger in metadata the proxy may persist to spend logs."""
+    metadata: Final = as_object(request.get("metadata"))
+    restore_map: Final = as_object(metadata.pop(RESTORE_MAP_KEY, None)) if metadata is not None else None
+    pairs: Final = (restore_map or {}).items()
+    return tuple((fake, original) for fake, original in pairs if isinstance(original, str))
+
+
+def _pseudo_message(text: str) -> JsonObject:
+    """A string leaf wrapped as a message, so it crosses the engine's single choke point like chat content."""
+    return {"role": "user", "content": text}
+
+
+def _json_escaped(pairs: Iterable[tuple[str, str]]) -> Iterator[tuple[str, str]]:
+    """The argument channels stream JSON source text, so their originals are restored JSON-escaped.
+
+    Placeholders need no escaping, so the fakes are the same on both sides.
+    """
+    for fake, original in pairs:
+        yield fake, json_escape(original)
+
+
+class PrivaiteGuardrail(CustomGuardrail):
+    def __init__(
+        self,
+        preset: str | None = None,
+        languages: str | None = None,
+        deanonymize: bool | str | None = True,
+        block_entities: str | Sequence[str] | None = None,
+        supported_event_hooks: Sequence[GuardrailEventHooks] = _REQUIRED_HOOKS,
+        **kwargs: Unpack[_CustomGuardrailOptions],  # noqa: LIT008  # kwargs-ok: forwarded verbatim to CustomGuardrail.__init__
+    ) -> None:
+        self.preset: Final = preset if preset in _PRESETS else "onnx"
+        self.languages: Final = languages or "en,fr"
+        self.deanonymize: Final = (
+            deanonymize if isinstance(deanonymize, bool) else str(deanonymize).strip().lower() not in _FALSE_STRINGS
+        )
+        self.block_entities: Final = _parse_block_entities(block_entities)
+        super().__init__(  # pyright: ignore[reportUnknownMemberType]  # CustomGuardrail.__init__ takes untyped **kwargs
+            supported_event_hooks=list(supported_event_hooks), **kwargs
+        )
+        if not isinstance(self.event_hook, Mode):
+            self.event_hook = _with_required_hooks(self.event_hook)
+        self._engine: PrivaiteEngine | None = None
+        self._engine_key: _EngineKey | None = None
+        self._lock: Final = asyncio.Lock()
+
+    def _languages(self) -> tuple[str, ...]:
+        languages: Final = tuple(language.strip() for language in self.languages.split(",") if language.strip())
+        return languages or ("en",)
+
+    async def _engine_for(self, languages: Sequence[str]) -> PrivaiteEngine:
+        key: Final = (self.preset, tuple(languages), self.deanonymize, self.block_entities)
+        if self._engine is not None and self._engine_key == key:
+            return self._engine
+        async with self._lock:
+            if self._engine is not None and self._engine_key == key:
+                return self._engine  # pragma: no cover
+            engine: Final = await build_engine(self.preset, languages, self.deanonymize, self.block_entities)
+            self._engine = engine
+            self._engine_key = key
+            return engine
+
+    async def _anonymize_request(self, data: JsonObject, engine: PrivaiteEngine) -> RestoreMapping | None:
+        """Anonymizes every scanned request surface in place under one shared mapping.
+
+        Chat messages go to the engine as they are. Every other string leaf is appended to
+        the same batch as a pseudo-message, so one process_request pass covers a crafted
+        request that carries several surfaces, under the same block gate and fail-closed policy.
+        """
+        messages: Final = as_array(data.get("messages")) or []
+        slots: Final = tuple(
+            itertools.chain(responses_input_slots(data), prompt_variable_slots(data), aux_field_slots(data))
+        )
+        if not messages and not slots:
+            return None
+        anonymized, mapping = await engine.process_request([*messages, *(_pseudo_message(text) for text, _ in slots)])
+        count: Final = len(messages)
+        if len(anonymized) != count + len(slots):
+            raise RuntimeError(
+                f"privaite returned {len(anonymized)} messages for {count + len(slots)}, "
+                "refusing to forward a request it did not scrub in full"
+            )
+        if messages:
+            messages[:] = anonymized[:count]
+        for (_, write), leaf_message in zip(slots, anonymized[count:], strict=True):
+            content = read_field(leaf_message, "content")
+            write(content if isinstance(content, str) else "")
+        return mapping
+
+    async def async_pre_call_hook(
+        self,
+        user_api_key_dict: UserAPIKeyAuth,
+        cache: DualCache,
+        data: JsonObject,
+        call_type: str,
+    ) -> JsonObject:
+        # metadata is caller controlled, so an incoming privaite_map is never trusted:
+        # the post-call hooks may only restore from a map this pre-call pass produced.
+        _drop_restore_map(data)
+        if (
+            not data.get("messages")
+            and not data.get("input")
+            and prompt_variables(data) is None
+            and not has_aux_fields(data)
+        ):
+            return data
+
+        engine: Final = await self._engine_for(self._languages())
+        blocked: Final = blocked_error()
+        try:
+            mapping: Final = await self._anonymize_request(data, engine)
+        except blocked as exc:
+            from fastapi import HTTPException
+
+            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
+
+        # A background request's result is fetched by a later poll the post-call hooks
+        # never see, so its map would stay in request state with no consumer.
+        if mapping is None or not self.deanonymize or mapping.is_empty or data.get("background"):
+            return data
+        restore_map: Final = dict(mapping.get_all_fakes())
+        metadata: Final = as_object(data.get("metadata"))
+        if metadata is None:
+            write_field(data, "metadata", {RESTORE_MAP_KEY: restore_map})
+        else:
+            metadata[RESTORE_MAP_KEY] = restore_map
+        return data
+
+    async def async_post_call_success_hook(
+        self,
+        data: JsonObject,
+        user_api_key_dict: UserAPIKeyAuth,
+        response: object,
+    ) -> object:
+        """Restores the original values in a copy of the response.
+
+        LiteLLM keeps the object it hands this hook for its response cache and its
+        success logs, and a cache write may serialize it after this hook ran. Restoring it
+        in place would put this caller's values in the cache under a key built from the
+        anonymized request, where another caller's identical-looking request hits them.
+        """
+        if not self.deanonymize:
+            return response
+        pairs: Final = _pop_restore_map(data)
+        if not pairs:
+            return response
+        mapping: Final = restore_mapping(pairs)
+        engine: Final = await self._engine_for(self._languages())
+        restored: Final = copy.deepcopy(response)
+        for choice in read_list(restored, "choices"):
+            message = read_field(choice, "message")
+            if message is not None:
+                await restore_message(message, engine, mapping)
+        await restore_responses_output(restored, engine, mapping)
+        return restored
+
+    async def async_post_call_streaming_iterator_hook(
+        self,
+        user_api_key_dict: UserAPIKeyAuth,
+        response: AsyncIterable[ModelResponseStream],
+        request_data: JsonObject,
+    ) -> AsyncGenerator[ModelResponseStream, None]:
+        """Restores the original values in copies of the chunks.
+
+        The stream wrapper keeps every chunk it yields and rebuilds the whole response from
+        them for the cache and the success logs once the stream ends, so the chunks
+        themselves keep their placeholders, as in the non-streaming hook.
+        """
+        pairs: Final = _pop_restore_map(request_data)
+        if not self.deanonymize or not pairs:
+            async for chunk in response:
+                yield chunk
+            return
+
+        restorer: Final = StreamRestorer(
+            restore_mapping(pairs),
+            restore_mapping(_json_escaped(pairs)),
+            restore_buffer_factory(),
+        )
+        async for chunk in response:
+            restored = copy.deepcopy(chunk)
+            restorer.restore_chunk(restored)
+            yield restored
+        for drained in restorer.drain():
+            yield drained
+
+    async def async_post_call_failure_hook(
+        self,
+        request_data: object,
+        original_exception: Exception,
+        user_api_key_dict: UserAPIKeyAuth,
+        traceback_str: str | None = None,
+    ) -> None:
+        # The success and streaming hooks never run when the call fails, so the map is
+        # dropped here too, before a failure spend-log can persist it.
+        _drop_restore_map(request_data)
diff --git a/litellm/proxy/guardrails/guardrail_hooks/privaite/request_walk.py b/litellm/proxy/guardrails/guardrail_hooks/privaite/request_walk.py
new file mode 100644
index 00000000000..c07b9aa539d
--- /dev/null
+++ b/litellm/proxy/guardrails/guardrail_hooks/privaite/request_walk.py
@@ -0,0 +1,237 @@
+import functools
+from collections.abc import Iterator, Sequence
+from types import MappingProxyType
+from typing import Final
+
+from .payload import (
+    JsonArray,
+    JsonObject,
+    Slot,
+    Write,
+    as_array,
+    as_object,
+    field_writer,
+    item_writer,
+)
+
+RESPONSES_OPAQUE_TYPES: Final = frozenset(
+    {
+        "reasoning",
+        "compaction",
+        "compaction_trigger",
+        "computer_call_output",
+        "image_generation_call",
+        "item_reference",
+        "mcp_list_tools",
+        "tool_search_call",
+        "tool_search_output",
+        "additional_tools",
+    }
+)
+"""Responses `input` items relayed whole: an opaque or binary payload, or a tool or
+pointer definition with no user text. Parity with the PrivAiTe gateway scrubber."""
+
+RESPONSES_DATA_FIELDS: Final = MappingProxyType(
+    {
+        "computer_call": ("action", "actions"),
+        "local_shell_call": ("action",),
+        "shell_call": ("action",),
+        "web_search_call": ("action",),
+        "apply_patch_call": ("operation",),
+        "file_search_call": ("queries", "results"),
+        "code_interpreter_call": ("code", "outputs"),
+        "program": ("code",),
+        "program_output": ("result",),
+    }
+)
+"""Responses items whose user data sits in known structured fields, walked leaf by leaf."""
+
+BINARY_PART_TYPES: Final = frozenset(
+    {
+        "input_image",
+        "input_file",
+        "input_audio",
+        "image",
+        "output_image",
+        "computer_screenshot",
+    }
+)
+"""Parts whose payload is base64 or a file id. Scrubbing one would corrupt it without
+removing anything a text detector can find."""
+
+GENERIC_TEXT_FIELDS: Final = ("output", "arguments", "input", "text", "reason")
+"""Text-bearing fields of an unknown Responses item. Every one is scanned, since an
+mcp_call carries both `arguments` and `output`."""
+
+
+def leaf(text: object, write: Write) -> Iterator[Slot]:
+    if isinstance(text, str) and text:
+        yield text, write
+
+
+def content_slots(content: object, write: Write) -> Iterator[Slot]:
+    """A role message's `content`: a bare string, or parts whose `text` and `refusal` carry user text."""
+    if isinstance(content, str):
+        yield from leaf(content, write)
+        return
+    parts: Final = as_array(content)
+    if parts is None:
+        return
+    for index, part in enumerate(parts):
+        if isinstance(part, str):
+            yield from leaf(part, item_writer(parts, index))
+            continue
+        fields = as_object(part)
+        if fields is None or fields.get("type") in BINARY_PART_TYPES:
+            continue
+        for name in ("text", "refusal"):
+            yield from leaf(fields.get(name), field_writer(fields, name))
+
+
+def data_slots(value: object, write: Write) -> Iterator[Slot]:
+    """Every string leaf of a tool payload, binary parts relayed whole and object keys never scanned.
+
+    The payload is caller controlled, so it is walked with an explicit stack rather than
+    by recursion. Children are pushed reversed, so leaves come out depth first.
+    """
+    pending: Final[list[tuple[object, Write]]] = [(value, write)]  # mutable-ok: local walk stack, never escapes
+    while pending:
+        node, write_back = pending.pop()
+        if isinstance(node, str):
+            yield from leaf(node, write_back)
+            continue
+        items = as_array(node)
+        if items is not None:
+            pending.extend((items[index], item_writer(items, index)) for index in range(len(items) - 1, -1, -1))
+            continue
+        fields = as_object(node)
+        if fields is not None and fields.get("type") not in BINARY_PART_TYPES:
+            pending.extend((child, field_writer(fields, key)) for key, child in reversed(fields.items()))
+
+
+def _named_data_slots(fields: JsonObject, names: Sequence[str]) -> Iterator[Slot]:
+    for name in names:
+        if fields.get(name) is not None:
+            yield from data_slots(fields[name], field_writer(fields, name))
+
+
+def _typed_item_slots(fields: JsonObject, item_type: str) -> Iterator[Slot] | None:
+    """The text of an item whose user data sits in a type-specific field, or None for an unknown shape."""
+    if item_type == "function_call" and isinstance(fields.get("arguments"), str):
+        return leaf(fields["arguments"], field_writer(fields, "arguments"))
+    if item_type == "custom_tool_call" and isinstance(fields.get("input"), str):
+        return leaf(fields["input"], field_writer(fields, "input"))
+    data_fields: Final = RESPONSES_DATA_FIELDS.get(item_type)
+    if data_fields is not None:
+        return _named_data_slots(fields, data_fields)
+    if item_type.endswith("_output") and "output" in fields:
+        return data_slots(fields["output"], field_writer(fields, "output"))
+    return None
+
+
+def _generic_item_slots(fields: JsonObject) -> Iterator[Slot]:
+    for name in GENERIC_TEXT_FIELDS:
+        yield from leaf(fields.get(name), field_writer(fields, name))
+    content: Final = fields.get("content")
+    if isinstance(content, str) or as_array(content) is not None:
+        yield from content_slots(content, field_writer(fields, "content"))
+
+
+def _responses_item_slots(items: JsonArray, index: int) -> Iterator[Slot]:
+    item: Final = items[index]
+    if isinstance(item, str):
+        yield from leaf(item, item_writer(items, index))
+        return
+    fields: Final = as_object(item)
+    if fields is None:
+        return
+    if "role" in fields and "content" in fields:
+        yield from content_slots(fields["content"], field_writer(fields, "content"))
+        return
+    declared_type: Final = fields.get("type")
+    item_type: Final = declared_type if isinstance(declared_type, str) else ""
+    if item_type in RESPONSES_OPAQUE_TYPES:
+        return
+    typed: Final = _typed_item_slots(fields, item_type)
+    yield from typed if typed is not None else _generic_item_slots(fields)
+
+
+def _write_top_level(data: JsonObject, key: str, value: str) -> None:
+    """Writes a top-level string field, and the copy of it in the proxy's request-body snapshot.
+
+    The proxy snapshots the body by shallow copy before the hooks run. A nested list or
+    dict is shared with that snapshot, but a top-level string is not, so without the
+    second write the original would stay in proxy_server_request.body.
+    """
+    data[key] = value  # rebind-ok: the pre-call hook contract rewrites the request in place
+    request: Final = as_object(data.get("proxy_server_request"))
+    body: Final = as_object(request.get("body")) if request is not None else None
+    if body is not None and key in body:
+        body[key] = value
+
+
+def top_level_writer(data: JsonObject, key: str) -> Write:
+    return functools.partial(_write_top_level, data, key)
+
+
+def responses_input_slots(data: JsonObject) -> Iterator[Slot]:
+    request_input: Final = data.get("input")
+    if isinstance(request_input, str):
+        yield from leaf(request_input, top_level_writer(data, "input"))
+        return
+    items: Final = as_array(request_input)
+    if items is None:
+        return
+    for index in range(len(items)):
+        yield from _responses_item_slots(items, index)
+
+
+def prompt_variables(data: JsonObject) -> JsonObject | None:
+    """Responses prompt-template variables, which carry user data. A string `prompt` is /v1/completions."""
+    prompt: Final = as_object(data.get("prompt"))
+    return as_object(prompt.get("variables")) if prompt is not None else None
+
+
+def prompt_variable_slots(data: JsonObject) -> Iterator[Slot]:
+    variables: Final = prompt_variables(data)
+    if variables is None:
+        return
+    for key, value in variables.items():
+        yield from data_slots(value, field_writer(variables, key))
+
+
+def has_aux_fields(data: JsonObject) -> bool:
+    """Whether the request carries a field scanned by `aux_field_slots`."""
+    prediction: Final = as_object(data.get("prediction"))
+    if prediction is not None and "content" in prediction:
+        return True
+    web_search: Final = as_object(data.get("web_search_options"))
+    if web_search is not None and "user_location" in web_search:
+        return True
+    suffix: Final = data.get("suffix")
+    if isinstance(suffix, str) and suffix:
+        return True
+    prompt: Final = data.get("prompt")
+    return (isinstance(prompt, str) and bool(prompt)) or bool(as_array(prompt))
+
+
+def aux_field_slots(data: JsonObject) -> Iterator[Slot]:
+    """Request text outside messages and input that LiteLLM forwards verbatim: chat
+    `prediction.content`, `web_search_options.user_location`, and the completions `suffix`
+    and `prompt`. A tokenized (integer array) prompt passes through unscanned."""
+    prediction: Final = as_object(data.get("prediction"))
+    if prediction is not None and prediction.get("content") is not None:
+        yield from data_slots(prediction["content"], field_writer(prediction, "content"))
+    web_search: Final = as_object(data.get("web_search_options"))
+    if web_search is not None and web_search.get("user_location") is not None:
+        yield from data_slots(web_search["user_location"], field_writer(web_search, "user_location"))
+    yield from leaf(data.get("suffix"), top_level_writer(data, "suffix"))
+    prompt: Final = data.get("prompt")
+    if isinstance(prompt, str):
+        yield from leaf(prompt, top_level_writer(data, "prompt"))
+        return
+    parts: Final = as_array(prompt)
+    if parts is None:
+        return
+    for index, part in enumerate(parts):
+        yield from leaf(part, item_writer(parts, index))
diff --git a/litellm/proxy/guardrails/guardrail_hooks/privaite/restore.py b/litellm/proxy/guardrails/guardrail_hooks/privaite/restore.py
new file mode 100644
index 00000000000..8ac3fdb54a7
--- /dev/null
+++ b/litellm/proxy/guardrails/guardrail_hooks/privaite/restore.py
@@ -0,0 +1,258 @@
+from __future__ import annotations
+
+import copy
+import json
+from collections.abc import Callable, Iterator
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Final, TypeAlias
+
+from .engine import PrivaiteEngine, RestoreBuffer, RestoreMapping
+from .payload import as_array, as_object, index_of, read_field, read_list, write_field
+
+if TYPE_CHECKING:
+    from litellm.types.utils import ModelResponseStream
+
+TEXT_FIELDS: Final = ("reasoning_content", "reasoning", "refusal")
+"""Text fields restored besides `content`. A refusal can quote the request, so it carries placeholders too."""
+
+DELTA_CHANNELS: Final = ("content", *TEXT_FIELDS, "audio", "tool_calls", "function_call")
+
+JSON_CHANNELS: Final = frozenset({"tool", "function_call"})
+"""Channels that stream JSON source text, where a restored original must arrive JSON-escaped."""
+
+
+def json_escape(value: str) -> str:
+    """The JSON string-literal encoding of value without its quotes, as privaite.streaming.buffer does."""
+    return json.dumps(value, ensure_ascii=False)[1:-1]
+
+
+async def _restore_json_tree(engine: PrivaiteEngine, value: object, mapping: RestoreMapping) -> object:
+    """Restores every string leaf of a parsed JSON document. Keys are left as they are, as in the core engine."""
+    if isinstance(value, str):
+        return await engine.process_response(value, mapping)
+    fields: Final = as_object(value)
+    if fields is not None:
+        return {key: await _restore_json_tree(engine, item, mapping) for key, item in fields.items()}
+    items: Final = as_array(value)
+    if items is not None:
+        return [await _restore_json_tree(engine, item, mapping) for item in items]
+    return value
+
+
+async def restore_arguments(engine: PrivaiteEngine, arguments: str, mapping: RestoreMapping) -> str:
+    """Restores a tool call's `arguments`, a JSON document inside a string.
+
+    Plain substitution would splice a raw quote, backslash or newline from the original
+    into a JSON string literal, and the client's parse of the arguments would fail. So
+    the restore runs on the parsed tree and re-encodes it, as PIIEngine does. When nothing
+    changes, or the string is not JSON, the plain restore is returned as is.
+    """
+    plain: Final = await engine.process_response(arguments, mapping)
+    if plain == arguments:
+        return arguments
+    try:
+        parsed: Final[object] = json.loads(arguments)  # pyright: ignore[reportAny]  # json.loads is untyped
+    except ValueError:
+        return plain
+    return json.dumps(await _restore_json_tree(engine, parsed, mapping), ensure_ascii=False)
+
+
+async def _restore_text(holder: object, name: str, engine: PrivaiteEngine, mapping: RestoreMapping) -> None:
+    value: Final = read_field(holder, name)
+    if isinstance(value, str) and value:
+        write_field(holder, name, await engine.process_response(value, mapping))
+
+
+async def _restore_call_arguments(holder: object, engine: PrivaiteEngine, mapping: RestoreMapping) -> None:
+    arguments: Final = read_field(holder, "arguments")
+    if isinstance(arguments, str) and arguments:
+        write_field(holder, "arguments", await restore_arguments(engine, arguments, mapping))
+
+
+async def restore_message(message: object, engine: PrivaiteEngine, mapping: RestoreMapping) -> None:
+    """Restores one response message: content, the text fields, the audio transcript, and tool-call arguments."""
+    for name in ("content", *TEXT_FIELDS):
+        await _restore_text(message, name, engine, mapping)
+    await _restore_text(read_field(message, "audio"), "transcript", engine, mapping)
+    for tool_call in read_list(message, "tool_calls"):
+        await _restore_call_arguments(read_field(tool_call, "function"), engine, mapping)
+    await _restore_call_arguments(read_field(message, "function_call"), engine, mapping)
+
+
+async def restore_responses_output(response: object, engine: PrivaiteEngine, mapping: RestoreMapping) -> None:
+    """Restores a Responses API result: output_text blocks and function_call arguments."""
+    for item in read_list(response, "output"):
+        for block in read_list(item, "content"):
+            await _restore_text(block, "text", engine, mapping)
+        await _restore_call_arguments(item, engine, mapping)
+
+
+@dataclass(frozen=True, slots=True)
+class Channel:
+    """One streamed segment of one choice. Each keeps its own restore buffer."""
+
+    kind: str
+    choice: int
+    tool_call: int = 0
+
+
+Restore: TypeAlias = Callable[[Channel, str, bool], str]
+
+
+def _append_text(holder: object, name: str, tail: str) -> None:
+    existing: Final = read_field(holder, name)
+    write_field(holder, name, (existing if isinstance(existing, str) else "") + tail)
+
+
+def _append_nested(delta: object, carrier: str, name: str, tail: str) -> None:
+    """Appends to delta.<carrier>.<name>, creating the carrier when a bare finish delta has none."""
+    holder: Final = read_field(delta, carrier)
+    if holder is None:
+        write_field(delta, carrier, {name: tail})
+        return
+    _append_text(holder, name, tail)
+
+
+def _append_tool_tail(delta: object, tool_call: int, tail: str) -> None:
+    """Appends an argument tail under its tool_call index, since the client reassembles arguments per index."""
+    calls: Final = as_array(read_field(delta, "tool_calls"))
+    added: Final = {"index": tool_call, "function": {"arguments": tail}}
+    if calls is None:
+        write_field(delta, "tool_calls", [added])
+        return
+    existing: Final = next((call for call in calls if index_of(call) == tool_call), None)
+    if existing is None:
+        calls.append(added)
+        return
+    _append_nested(existing, "function", "arguments", tail)
+
+
+def append_tail(delta: object, channel: Channel, tail: str) -> None:
+    """Appends a restore buffer's held tail onto a streamed delta."""
+    match channel.kind:
+        case "tool":
+            _append_tool_tail(delta, channel.tool_call, tail)
+        case "function_call":
+            _append_nested(delta, "function_call", "arguments", tail)
+        case "audio":
+            _append_nested(delta, "audio", "transcript", tail)
+        case _:
+            _append_text(delta, channel.kind, tail)
+
+
+def _restore_delta_audio(delta: object, choice: int, finished: bool, restore: Restore) -> None:
+    """Restores a streamed audio transcript, creating the carrier on a finish delta so a held tail is kept."""
+    audio: Final = read_field(delta, "audio")
+    transcript: Final = read_field(audio, "transcript")
+    fragment: Final = transcript if isinstance(transcript, str) else ""
+    if not fragment and not finished:
+        return
+    restored: Final = restore(Channel("audio", choice), fragment, finished)
+    if audio is not None:
+        write_field(audio, "transcript", restored)
+    elif restored:
+        write_field(delta, "audio", {"transcript": restored})
+
+
+def restore_delta(delta: object, choice: int, finished: bool, restore: Restore) -> None:
+    """Restores one streamed delta: content, the text fields, the audio transcript and argument fragments."""
+    content: Final = read_field(delta, "content")
+    text: Final = content if isinstance(content, str) else ""
+    restored: Final = restore(Channel("content", choice), text, finished)
+    # A fragment held back whole must not stay visible raw, but a terminal None with
+    # nothing held back keeps its None rather than becoming "".
+    if text or restored:
+        write_field(delta, "content", restored)
+    for name in TEXT_FIELDS:
+        value = read_field(delta, name)
+        if isinstance(value, str) and (value or finished):
+            write_field(delta, name, restore(Channel(name, choice), value, finished))
+    _restore_delta_audio(delta, choice, finished, restore)
+    for tool_call in read_list(delta, "tool_calls"):
+        function = read_field(tool_call, "function")
+        arguments = read_field(function, "arguments")
+        if isinstance(arguments, str) and arguments:
+            channel = Channel("tool", choice, index_of(tool_call))
+            write_field(function, "arguments", restore(channel, arguments, finished))
+    function_call: Final = read_field(delta, "function_call")
+    call_arguments: Final = read_field(function_call, "arguments")
+    if isinstance(call_arguments, str) and call_arguments:
+        write_field(function_call, "arguments", restore(Channel("function_call", choice), call_arguments, finished))
+
+
+def drain_chunk(template: ModelResponseStream, channel: Channel, tail: str) -> ModelResponseStream:
+    """A trailing chunk carrying one held tail, for a stream the provider closed without a finish_reason.
+
+    It is a copy of the last chunk of that choice, so id, model and object match the rest
+    of the stream, reduced to that one choice with its delta cleared and no usage.
+    """
+    chunk: Final = copy.deepcopy(template)
+    choice: Final = next(item for item in read_list(chunk, "choices") if index_of(item) == channel.choice)
+    write_field(chunk, "choices", [choice])
+    if read_field(chunk, "usage") is not None:
+        write_field(chunk, "usage", None)
+    write_field(choice, "finish_reason", None)
+    delta: Final = read_field(choice, "delta")
+    for name in DELTA_CHANNELS:
+        if read_field(delta, name) is not None:
+            write_field(delta, name, None)
+    append_tail(delta, channel, tail)
+    return chunk
+
+
+class StreamRestorer:
+    """Restore state for one streamed response.
+
+    With n>1 the provider interleaves chunks of different choices, and tool-call arguments
+    stream as fragments per tool_call index, so every channel keeps its own buffer and a
+    placeholder split across chunks reassembles without mixing channels. A held tail is
+    flushed onto the chunk carrying finish_reason, or, when the provider closes the stream
+    without one, drained as trailing chunks. It is never dropped.
+    """
+
+    def __init__(
+        self,
+        mapping: RestoreMapping,
+        json_mapping: RestoreMapping,
+        make_buffer: Callable[[RestoreMapping], RestoreBuffer],
+    ) -> None:
+        self._mapping: Final = mapping
+        self._json_mapping: Final = json_mapping
+        self._make_buffer: Final = make_buffer
+        self._buffers: Final[dict[Channel, RestoreBuffer]] = {}  # mutable-ok: per-stream restore state
+        self._last_chunk: Final[dict[int, ModelResponseStream]] = {}  # mutable-ok: per-stream restore state
+
+    def restore(self, channel: Channel, text: str, finished: bool) -> str:
+        if channel not in self._buffers:
+            mapping: Final = self._json_mapping if channel.kind in JSON_CHANNELS else self._mapping
+            self._buffers[channel] = self._make_buffer(mapping)
+        restored: Final = self._buffers[channel].feed(text) if text else ""
+        if not finished:
+            return restored
+        return restored + self._buffers.pop(channel).flush()
+
+    def _close(self, choice: int | None) -> tuple[tuple[Channel, str], ...]:
+        """Flushes and drops every buffer still open for a choice, or for all of them; returns the non-empty tails."""
+        closing: Final = tuple(channel for channel in self._buffers if choice is None or channel.choice == choice)
+        flushed: Final = tuple((channel, self._buffers.pop(channel).flush()) for channel in closing)
+        return tuple((channel, tail) for channel, tail in flushed if tail)
+
+    def restore_chunk(self, chunk: ModelResponseStream) -> None:
+        """Restores every choice of a chunk, and appends the tails of the channels a finish delta does not carry."""
+        for choice in read_list(chunk, "choices"):
+            delta = read_field(choice, "delta")
+            if delta is None:
+                continue
+            index = index_of(choice)
+            self._last_chunk[index] = chunk
+            finished = read_field(choice, "finish_reason") is not None
+            restore_delta(delta, index, finished, self.restore)
+            if finished:
+                for channel, tail in self._close(index):
+                    append_tail(delta, channel, tail)
+
+    def drain(self) -> Iterator[ModelResponseStream]:
+        for channel, tail in self._close(None):
+            template = self._last_chunk.get(channel.choice)
+            if template is not None:
+                yield drain_chunk(template, channel, tail)
diff --git a/litellm/proxy/spend_tracking/spend_tracking_utils.py b/litellm/proxy/spend_tracking/spend_tracking_utils.py
index 94b079163e9..2e2beb4e5ab 100644
--- a/litellm/proxy/spend_tracking/spend_tracking_utils.py
+++ b/litellm/proxy/spend_tracking/spend_tracking_utils.py
@@ -1210,7 +1210,7 @@ def _get_messages_for_spend_logs_payload(
     return "{}"
 
 
-_SENSITIVE_REQUEST_BODY_KEYS: Final = frozenset({"secret_fields"})
+_SENSITIVE_REQUEST_BODY_KEYS: Final = frozenset({"secret_fields", "privaite_map"})
 _REQUEST_BODY_CREDENTIAL_MASKER: Final = SensitiveDataMasker(extra_sensitive_patterns=frozenset({"apikey"}))
 _TOOL_INPUT_BLOCK_TYPES: Final = frozenset({"tool_use", "server_tool_use", "mcp_tool_use"})
 _TOOL_OUTPUT_BLOCK_TYPES: Final = frozenset({"tool_result", "mcp_tool_result", "function_call_output"})
diff --git a/litellm/types/guardrails.py b/litellm/types/guardrails.py
index ff063d2e233..62c2f9b2857 100644
--- a/litellm/types/guardrails.py
+++ b/litellm/types/guardrails.py
@@ -50,6 +50,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter impor
 from litellm.types.proxy.guardrails.guardrail_hooks.ovalix import (
     OvalixGuardrailConfigModel,
 )
+from litellm.types.proxy.guardrails.guardrail_hooks.privaite import (
+    PrivaiteGuardrailConfigModel,
+)
 from litellm.types.proxy.guardrails.guardrail_hooks.promptguard import (
     PromptGuardConfigModel,
 )
@@ -159,6 +162,7 @@ class SupportedGuardrailIntegrations(Enum):
     AIRIA = "airia"
     WINGBACK = "wingback"
     DECISION_MODEL = "decision_model"
+    PRIVAITE = "privaite"
 
 
 class Role(Enum):
@@ -1400,6 +1404,7 @@ class LitellmParams(  # pyright: ignore[reportIncompatibleVariableOverride]  # o
     SingulrGuardrailConfigModel,
     Agent365GuardrailConfigModel,
     WingbackGuardrailConfigModel,
+    PrivaiteGuardrailConfigModel,
 ):
     guardrail: str = Field(description="The type of guardrail integration to use")
     decision_model: str | None = Field(
diff --git a/litellm/types/proxy/guardrails/guardrail_hooks/privaite.py b/litellm/types/proxy/guardrails/guardrail_hooks/privaite.py
new file mode 100644
index 00000000000..98db0c9211a
--- /dev/null
+++ b/litellm/types/proxy/guardrails/guardrail_hooks/privaite.py
@@ -0,0 +1,31 @@
+from pydantic import Field
+
+from .base import GuardrailConfigModel
+
+
+class PrivaiteGuardrailConfigModel(GuardrailConfigModel):  # pyright: ignore[reportMissingTypeArgument]  # parameterizing adds an optional_params field to the UI form
+    preset: str | None = Field(
+        default="onnx",
+        description=("Detection preset: 'onnx' (full suite, also detects secrets) or 'light' (faster, Presidio only)."),
+    )
+    languages: str | None = Field(
+        default="en,fr",
+        description="Comma-separated spaCy languages used for detection.",
+    )
+    deanonymize: bool | None = Field(
+        default=True,
+        description="Restore the original PII values in the response.",
+    )
+    block_entities: str | tuple[str, ...] | None = Field(
+        default=None,
+        description=(
+            "PII types to reject outright instead of masking: a list, or a "
+            "comma-separated string, e.g. ['US_SSN', 'CREDIT_CARD']. A request "
+            "containing any listed type is refused with a 400 and nothing is "
+            "forwarded. Omitted (the default) masks everything and forwards it."
+        ),
+    )
+
+    @staticmethod
+    def ui_friendly_name() -> str:
+        return "PrivAiTe"
diff --git a/tests/unit/proxy/guardrails/guardrail_hooks/test_privaite.py b/tests/unit/proxy/guardrails/guardrail_hooks/test_privaite.py
new file mode 100644
index 00000000000..5a19ce91fe2
--- /dev/null
+++ b/tests/unit/proxy/guardrails/guardrail_hooks/test_privaite.py
@@ -0,0 +1,1460 @@
+"""
+Unit tests for the PrivAiTe guardrail provider.
+
+The PrivAiTe engine lives in the external 'privaite' package, which is NOT a
+dependency of LiteLLM. These tests therefore install a fake 'privaite' package
+into sys.modules (an autouse fixture) so the guardrail's lazy imports resolve to
+fakes. That means the suite runs exactly the way LiteLLM CI runs it: with the
+real 'privaite' package absent. The fakes let us exercise the guardrail's own
+wiring end to end (registry, engine caching, pre/post/streaming hooks) without
+re-testing PrivAiTe's detection internals.
+"""
+
+import json
+import sys
+import types
+
+import pytest
+
+# fake placeholder <-> original, the way PrivAiTe maps values.
+_FAKES = {"<PERSON_1>": "Marie Dupont", "<EMAIL_ADDRESS_1>": "marie@acme.com"}
+_REALS = {original: fake for fake, original in _FAKES.items()}
+
+
+class _FakeMapping:
+    """Stand-in for privaite.pii.mapping.PIIMapping."""
+
+    def __init__(self, fakes=None):
+        self._fakes = dict(fakes or {})
+
+    @property
+    def is_empty(self):
+        return not self._fakes
+
+    def get_all_fakes(self):
+        return dict(self._fakes)
+
+    def add(self, original, fake, _entity_type):
+        self._fakes[fake] = original
+
+
+class _FakePIIBlockedError(Exception):
+    """Stand-in for privaite.pii.engine.PIIBlockedError. Names TYPES, not values."""
+
+    def __init__(self, entity_types):
+        self.entity_types = sorted(entity_types)
+        super().__init__("request blocked: contains disallowed PII type(s): " + ", ".join(self.entity_types))
+
+
+class _FakeEngine:
+    """Stand-in for privaite.pii.engine.PIIEngine.
+
+    Scrubs the known reals on the way out, restores the known fakes on the way
+    back. ``raise_oserror_until`` lets a test force the spaCy-download retry path
+    in the guardrail's _engine_for by failing the first N initialize() calls. If
+    ``config.block_entities`` names a type present in the text, process_request
+    raises _FakePIIBlockedError, exactly as the real engine would.
+    """
+
+    raise_oserror_until = 0
+    oserror_message = "[E050] Can't find model 'en_core_web_lg'."
+    init_calls = 0
+
+    def __init__(self, config):
+        self.config = config
+
+    async def initialize(self):
+        type(self).init_calls += 1
+        if self.init_calls <= type(self).raise_oserror_until:
+            raise OSError(type(self).oserror_message)
+
+    def _scrub(self, content, mapping):
+        if isinstance(content, str):
+            for real, fake in _REALS.items():
+                if real in content:
+                    content = content.replace(real, fake)
+                    mapping.add(real, fake, "PII")
+            return content
+        if isinstance(content, list):
+            out = []
+            for part in content:
+                if isinstance(part, str):
+                    out.append(self._scrub(part, mapping))
+                elif isinstance(part, dict) and isinstance(part.get("text"), str):
+                    out.append({**part, "text": self._scrub(part["text"], mapping)})
+                else:
+                    out.append(part)
+            return out
+        return content
+
+    async def process_request(self, messages):
+        mapping = _FakeMapping()
+        anonymized = []
+        for message in messages:
+            new_message = dict(message)
+            new_message["content"] = self._scrub(message.get("content", ""), mapping)
+            anonymized.append(new_message)
+        blocked = set(getattr(self.config, "block_entities", None) or [])
+        if blocked:
+            # placeholder "<EMAIL_ADDRESS_1>" -> type "EMAIL_ADDRESS"
+            detected = {fake.strip("<>").rsplit("_", 1)[0] for fake in mapping.get_all_fakes()}
+            hit = detected & blocked
+            if hit:
+                raise _FakePIIBlockedError(hit)
+        return anonymized, mapping
+
+    async def process_response(self, text, _mapping):
+        for fake, original in _FAKES.items():
+            text = text.replace(fake, original)
+        return text
+
+
+class _FakeStreamingDeAnonymizer:
+    """Stand-in for privaite.streaming.buffer.StreamingDeAnonymizer.
+
+    Stateful like the real one: it buffers a trailing "<...."-without-">" fragment
+    so a placeholder split across chunks reassembles, which is what makes per-choice
+    buffer isolation observable in tests.
+    """
+
+    def __init__(self, mapping):
+        self._fakes = mapping.get_all_fakes()
+        self._buf = ""
+
+    def feed(self, text):
+        self._buf += text
+        for fake, original in self._fakes.items():
+            self._buf = self._buf.replace(fake, original)
+        lt = self._buf.rfind("<")
+        if lt != -1 and ">" not in self._buf[lt:]:
+            emit, self._buf = self._buf[:lt], self._buf[lt:]
+        else:
+            emit, self._buf = self._buf, ""
+        return emit
+
+    def flush(self):
+        out, self._buf = self._buf, ""
+        return out
+
+
+class _Config:
+    """Generic stand-in for the privaite.config.schema dataclasses."""
+
+    def __init__(self, **kwargs):
+        self.__dict__.update(kwargs)
+
+
+class _PIIConfig(_Config):
+    """PIIConfig stand-in. ``model_fields`` mirrors a modern privaite that declares
+    block_entities, so the guardrail's fail-closed guard sees the field as present.
+    A test can override model_fields to simulate an older privaite."""
+
+    model_fields = {"preset": None, "languages": None, "block_entities": None}
+
+
+@pytest.fixture(autouse=True)
+def _fake_privaite_package(monkeypatch):
+    """Install a fake 'privaite' (and 'spacy') package tree so the guardrail's
+    lazy imports resolve without the real packages installed, mirroring CI."""
+    _FakeEngine.raise_oserror_until = 0
+    _FakeEngine.oserror_message = "[E050] Can't find model 'en_core_web_lg'."
+    _FakeEngine.init_calls = 0
+
+    def _module(name, **attrs):
+        mod = types.ModuleType(name)
+        for key, value in attrs.items():
+            setattr(mod, key, value)
+        return mod
+
+    schema = _module(
+        "privaite.config.schema",
+        AnonymizationConfig=_Config,
+        DeanonymizationConfig=_Config,
+        DetectorsConfig=_Config,
+        PIIConfig=_PIIConfig,
+        PresidioDetectorConfig=_Config,
+    )
+    mapping = _module("privaite.pii.mapping", PIIMapping=_FakeMapping)
+    engine = _module("privaite.pii.engine", PIIEngine=_FakeEngine, PIIBlockedError=_FakePIIBlockedError)
+    buffer = _module("privaite.streaming.buffer", StreamingDeAnonymizer=_FakeStreamingDeAnonymizer)
+    config_pkg = _module("privaite.config", schema=schema)
+    pii_pkg = _module("privaite.pii", mapping=mapping, engine=engine)
+    streaming_pkg = _module("privaite.streaming", buffer=buffer)
+    privaite_pkg = _module("privaite", config=config_pkg, pii=pii_pkg, streaming=streaming_pkg)
+    spacy_cli = _module("spacy.cli", download=lambda _model: None)
+    spacy_pkg = _module("spacy", cli=spacy_cli)
+
+    for name, mod in {
+        "privaite": privaite_pkg,
+        "privaite.config": config_pkg,
+        "privaite.config.schema": schema,
+        "privaite.pii": pii_pkg,
+        "privaite.pii.mapping": mapping,
+        "privaite.pii.engine": engine,
+        "privaite.streaming": streaming_pkg,
+        "privaite.streaming.buffer": buffer,
+        "spacy": spacy_pkg,
+        "spacy.cli": spacy_cli,
+    }.items():
+        monkeypatch.setitem(sys.modules, name, mod)
+    yield
+
+
+def _make_guardrail(**kwargs):
+    from litellm.proxy.guardrails.guardrail_hooks.privaite import PrivaiteGuardrail
+
+    return PrivaiteGuardrail(guardrail_name="privaite-test", **kwargs)
+
+
+async def _collect(aiterator):
+    return [chunk async for chunk in aiterator]
+
+
+def test_registry_and_enum_wiring():
+    from litellm.proxy.guardrails.guardrail_hooks.privaite import (
+        PrivaiteGuardrail,
+        guardrail_class_registry,
+        guardrail_initializer_registry,
+    )
+    from litellm.types.guardrails import SupportedGuardrailIntegrations
+
+    assert SupportedGuardrailIntegrations.PRIVAITE.value == "privaite"
+    assert "privaite" in guardrail_initializer_registry
+    assert guardrail_class_registry["privaite"] is PrivaiteGuardrail
+
+
+def test_config_model_ui_name():
+    from litellm.types.proxy.guardrails.guardrail_hooks.privaite import (
+        PrivaiteGuardrailConfigModel,
+    )
+
+    assert PrivaiteGuardrailConfigModel.ui_friendly_name() == "PrivAiTe"
+    fields = PrivaiteGuardrailConfigModel.model_fields
+    assert fields["preset"].default == "onnx"
+    assert fields["languages"].default == "en,fr"
+    assert fields["deanonymize"].default is True
+
+
+def test_initialize_guardrail_registers_callback(monkeypatch):
+    import litellm
+    from litellm.proxy.guardrails.guardrail_hooks.privaite import (
+        PrivaiteGuardrail,
+        initialize_guardrail,
+    )
+    from litellm.types.guardrails import LitellmParams
+
+    added = []
+    monkeypatch.setattr(
+        litellm.logging_callback_manager,
+        "add_litellm_callback",
+        lambda cb: added.append(cb),
+    )
+
+    litellm_params = LitellmParams(
+        guardrail="privaite",
+        mode="pre_call",
+        default_on=False,
+        preset="light",
+        languages="en",
+        deanonymize=True,
+    )
+    callback = initialize_guardrail(litellm_params, {"guardrail_name": "privaite"})
+
+    assert isinstance(callback, PrivaiteGuardrail)
+    assert callback.preset == "light"
+    assert added == [callback]
+
+
+def test_init_normalizes_params():
+    # invalid preset falls back to onnx; "false"-ish strings disable deanonymize.
+    gr = _make_guardrail(preset="bogus", languages="en, fr ,", deanonymize="false")
+    assert gr.preset == "onnx"
+    assert gr.deanonymize is False
+    assert gr._languages() == ("en", "fr")
+
+    # an explicit valid preset and bool are kept; empty languages fall back to
+    # the __init__ default of "en,fr".
+    gr2 = _make_guardrail(preset="light", languages="", deanonymize=True)
+    assert gr2.preset == "light"
+    assert gr2.deanonymize is True
+    assert gr2._languages() == ("en", "fr")
+
+    # a non-empty but content-free languages string falls back to ["en"].
+    assert _make_guardrail(languages=", ,")._languages() == ("en",)
+
+
+def test_event_hook_always_includes_pre_and_post():
+    # A `mode: post_call` config must NOT disable pre_call anonymization.
+    gr = _make_guardrail(event_hook="post_call")
+    assert "pre_call" in gr.event_hook and "post_call" in gr.event_hook
+    # A list missing post_call gets it added (and vice-versa).
+    gr2 = _make_guardrail(event_hook=["pre_call"])
+    assert "pre_call" in gr2.event_hook and "post_call" in gr2.event_hook
+    # The default (no event_hook configured) still runs both hooks.
+    gr3 = _make_guardrail()
+    assert "pre_call" in gr3.event_hook and "post_call" in gr3.event_hook
+
+
+@pytest.mark.asyncio
+async def test_engine_missing_privaite_recommends_safe_minimum(monkeypatch: pytest.MonkeyPatch) -> None:
+    guardrail = _make_guardrail()
+    monkeypatch.setitem(sys.modules, "privaite.config.schema", None)
+
+    with pytest.raises(RuntimeError, match=r"pip install 'privaite>=0\.7\.2'"):
+        await guardrail._engine_for(["en"])
+
+
+@pytest.mark.asyncio
+async def test_engine_is_cached():
+    gr = _make_guardrail()
+    engine_a = await gr._engine_for(["en"])
+    engine_b = await gr._engine_for(["en"])
+    assert engine_a is engine_b
+
+
+@pytest.mark.asyncio
+async def test_engine_downloads_spacy_models_on_oserror():
+    _FakeEngine.raise_oserror_until = 1  # first initialize() raises, retry succeeds
+    gr = _make_guardrail(languages="en,fr")
+    engine = await gr._engine_for(["en", "fr"])
+    assert isinstance(engine, _FakeEngine)
+    assert _FakeEngine.init_calls == 2  # failed once, then succeeded after download
+
+
+@pytest.mark.asyncio
+async def test_engine_does_not_retry_non_model_oserror():
+    # Only the missing-spaCy-model OSError may trigger the download retry;
+    # a disk/permission/network OSError must propagate untouched.
+    _FakeEngine.raise_oserror_until = 1
+    _FakeEngine.oserror_message = "disk full"
+    gr = _make_guardrail()
+    with pytest.raises(OSError, match="disk full"):
+        await gr._engine_for(["en"])
+    assert _FakeEngine.init_calls == 1  # no retry, no download
+
+
+@pytest.mark.asyncio
+async def test_pre_call_no_messages_is_passthrough():
+    gr = _make_guardrail()
+    data = {"messages": []}
+    assert await gr.async_pre_call_hook(None, None, data, "completion") is data
+
+
+@pytest.mark.asyncio
+async def test_pre_call_anonymizes_text_and_stashes_map():
+    gr = _make_guardrail()
+    messages = [{"role": "user", "content": "Email Marie Dupont at marie@acme.com"}]
+    data = {"messages": messages}
+    out = await gr.async_pre_call_hook(None, None, data, "completion")
+
+    # Mutated IN PLACE (same list object), so the proxy's shallow body snapshot
+    # ends up pointing at the anonymized messages, not the original raw-PII ones.
+    assert out["messages"] is messages
+    content = messages[0]["content"]
+    assert "Marie Dupont" not in content
+    assert "marie@acme.com" not in content
+    assert "<PERSON_1>" in content and "<EMAIL_ADDRESS_1>" in content
+    assert out["metadata"]["privaite_map"] == _FAKES
+
+
+@pytest.mark.asyncio
+async def test_pre_call_without_deanonymize_does_not_stash():
+    gr = _make_guardrail(deanonymize=False)
+    data = {"messages": [{"role": "user", "content": "Hi Marie Dupont"}]}
+    out = await gr.async_pre_call_hook(None, None, data, "completion")
+    assert "<PERSON_1>" in out["messages"][0]["content"]
+    assert "privaite_map" not in out.get("metadata", {})
+
+
+@pytest.mark.asyncio
+async def test_pre_call_without_pii_does_not_stash():
+    gr = _make_guardrail()
+    data = {"messages": [{"role": "user", "content": "no personal data here"}]}
+    out = await gr.async_pre_call_hook(None, None, data, "completion")
+    assert "privaite_map" not in out.get("metadata", {})
+
+
+@pytest.mark.asyncio
+async def test_pre_call_background_request_anonymizes_but_does_not_stash():
+    # Background results are fetched by a later poll the post-call hooks never
+    # see, so the fake->original map must not persist in request state.
+    gr = _make_guardrail()
+    data = {
+        "input": "Email Marie Dupont at marie@acme.com",
+        "background": True,
+    }
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+    assert "Marie Dupont" not in out["input"]
+    assert "<PERSON_1>" in out["input"]
+    assert "privaite_map" not in out.get("metadata", {})
+
+
+@pytest.mark.asyncio
+async def test_pre_call_clears_client_supplied_map():
+    # metadata is caller-controlled: a client-supplied privaite_map must be
+    # dropped, so it can never drive post-call restoration of model output.
+    gr = _make_guardrail()
+    data = {
+        "messages": [{"role": "user", "content": "no personal data here"}],
+        "metadata": {"privaite_map": {"<PERSON_1>": "attacker-chosen text"}},
+    }
+    out = await gr.async_pre_call_hook(None, None, data, "completion")
+    assert "privaite_map" not in out["metadata"]
+
+
+@pytest.mark.asyncio
+async def test_post_call_restores_text_and_tool_call_args():
+    gr = _make_guardrail()
+    data = {"metadata": {"privaite_map": _FAKES}}
+
+    tool_call = types.SimpleNamespace(function=types.SimpleNamespace(arguments='{"to": "<EMAIL_ADDRESS_1>"}'))
+    # a tool_call with no function must be skipped, not crash.
+    empty_tool_call = types.SimpleNamespace(function=None)
+    message = types.SimpleNamespace(
+        content="Sending to <PERSON_1>",
+        tool_calls=[tool_call, empty_tool_call],
+        function_call=types.SimpleNamespace(arguments='{"name": "<PERSON_1>"}'),
+    )
+    response = types.SimpleNamespace(
+        choices=[
+            types.SimpleNamespace(message=None),  # choice with no message is skipped
+            types.SimpleNamespace(message=message),
+        ]
+    )
+
+    out = await gr.async_post_call_success_hook(data, None, response)
+
+    msg = out.choices[1].message
+    assert msg.content == "Sending to Marie Dupont"
+    assert msg.tool_calls[0].function.arguments == '{"to": "marie@acme.com"}'
+    assert msg.function_call.arguments == '{"name": "Marie Dupont"}'
+    # the map is consumed (popped) so it cannot be persisted to spend logs.
+    assert "privaite_map" not in data["metadata"]
+
+
+@pytest.mark.asyncio
+async def test_post_call_deanonymize_false_skips_restore():
+    gr = _make_guardrail(deanonymize=False)
+    data = {"metadata": {"privaite_map": _FAKES}}
+    message = types.SimpleNamespace(content="Sending to <PERSON_1>", tool_calls=None, function_call=None)
+    response = types.SimpleNamespace(choices=[types.SimpleNamespace(message=message)])
+
+    out = await gr.async_post_call_success_hook(data, None, response)
+    assert out.choices[0].message.content == "Sending to <PERSON_1>"
+
+
+@pytest.mark.asyncio
+async def test_post_call_without_map_is_passthrough():
+    gr = _make_guardrail()
+    response = types.SimpleNamespace(choices=[])
+    assert await gr.async_post_call_success_hook({}, None, response) is response
+
+
+@pytest.mark.asyncio
+async def test_streaming_restores_content():
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    async def _source():
+        # a choice with no delta must be skipped, not crash.
+        yield types.SimpleNamespace(choices=[types.SimpleNamespace(delta=None, finish_reason=None)])
+        yield types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    delta=types.SimpleNamespace(content="Hi <PERSON_1>"),
+                    finish_reason=None,
+                )
+            ]
+        )
+        yield types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    delta=types.SimpleNamespace(content=""),
+                    finish_reason="stop",
+                )
+            ]
+        )
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    assert chunks[1].choices[0].delta.content == "Hi Marie Dupont"
+
+
+@pytest.mark.asyncio
+async def test_streaming_terminal_none_content_stays_none():
+    # A provider finish chunk commonly carries finish_reason with content=None;
+    # with nothing held back, restoration must not mutate that None into "".
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    async def _source():
+        yield types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    delta=types.SimpleNamespace(content="Hi <PERSON_1>"),
+                    finish_reason=None,
+                )
+            ]
+        )
+        yield types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    delta=types.SimpleNamespace(content=None),
+                    finish_reason="stop",
+                )
+            ]
+        )
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    assert chunks[0].choices[0].delta.content == "Hi Marie Dupont"
+    assert chunks[1].choices[0].delta.content is None
+
+
+@pytest.mark.asyncio
+async def test_streaming_passthrough_when_disabled():
+    gr = _make_guardrail(deanonymize=False)
+
+    async def _source():
+        yield types.SimpleNamespace(choices=[])
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), {}))
+    assert len(chunks) == 1
+
+
+@pytest.mark.asyncio
+async def test_streaming_isolates_buffers_per_choice():
+    # With n>1 the provider interleaves single-choice chunks for different
+    # indices. A placeholder split across choice 0's chunks must not be
+    # corrupted by choice 1's interleaved bytes (one buffer per choice index).
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    def _chunk(index, content, finish=None):
+        return types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    index=index,
+                    delta=types.SimpleNamespace(content=content),
+                    finish_reason=finish,
+                )
+            ]
+        )
+
+    async def _source():
+        yield _chunk(0, "Hi <PER")  # placeholder starts, held in choice-0 buffer
+        yield _chunk(1, "Bye")  # interleaved choice-1 content
+        yield _chunk(0, "SON_1>", finish="stop")  # completes choice-0 placeholder
+        yield _chunk(1, "", finish="stop")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+
+    restored = {}
+    for chunk in chunks:
+        for choice in chunk.choices:
+            restored.setdefault(choice.index, "")
+            if choice.delta.content:
+                restored[choice.index] += choice.delta.content
+
+    assert restored[0] == "Hi Marie Dupont"
+    assert restored[1] == "Bye"
+
+
+@pytest.mark.asyncio
+async def test_streaming_restores_reasoning_content():
+    # a reasoning model echoes a placeholder in its streamed reasoning trace; it
+    # must be de-anonymized like content.
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    def _chunk(reasoning, finish=None):
+        return types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    index=0,
+                    delta=types.SimpleNamespace(
+                        content=None, tool_calls=None, function_call=None, reasoning_content=reasoning
+                    ),
+                    finish_reason=finish,
+                )
+            ]
+        )
+
+    async def _source():
+        yield _chunk("the user is <PERSON_1>")
+        yield _chunk(None, finish="stop")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    trace = "".join(
+        getattr(choice.delta, "reasoning_content", None) or "" for chunk in chunks for choice in chunk.choices
+    )
+    assert "Marie Dupont" in trace
+    assert "<PERSON_1>" not in trace
+
+
+@pytest.mark.asyncio
+async def test_streaming_restores_tool_and_function_call_arguments():
+    # Streamed tool-call and legacy function_call argument fragments must be
+    # de-anonymized too, with a placeholder split across chunks reassembled per
+    # tool_call index. A tool_call with no function is skipped, not crashed.
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    async def _source():
+        yield types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    index=0,
+                    delta=types.SimpleNamespace(
+                        content=None,
+                        tool_calls=[
+                            types.SimpleNamespace(
+                                index=0,
+                                function=types.SimpleNamespace(arguments='{"to": "<EMAIL_ADDRESS'),
+                            ),
+                            types.SimpleNamespace(index=1, function=None),
+                        ],
+                        function_call=types.SimpleNamespace(arguments='{"n": "<PER'),
+                    ),
+                    finish_reason=None,
+                )
+            ]
+        )
+        yield types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    index=0,
+                    delta=types.SimpleNamespace(
+                        content=None,
+                        tool_calls=[
+                            types.SimpleNamespace(
+                                index=0,
+                                function=types.SimpleNamespace(arguments='_1>"}'),
+                            )
+                        ],
+                        function_call=types.SimpleNamespace(arguments='SON_1>"}'),
+                    ),
+                    finish_reason="stop",
+                )
+            ]
+        )
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+
+    tool_args = "".join(
+        tc.function.arguments
+        for chunk in chunks
+        for choice in chunk.choices
+        for tc in (choice.delta.tool_calls or [])
+        if tc.function is not None and tc.function.arguments
+    )
+    fc_args = "".join(
+        choice.delta.function_call.arguments
+        for chunk in chunks
+        for choice in chunk.choices
+        if choice.delta.function_call is not None and choice.delta.function_call.arguments
+    )
+    assert tool_args == '{"to": "marie@acme.com"}'
+    assert fc_args == '{"n": "Marie Dupont"}'
+
+
+# --- Responses API (/v1/responses) ---
+
+
+@pytest.mark.asyncio
+async def test_pre_call_anonymizes_responses_string_input_and_fixes_snapshot():
+    gr = _make_guardrail()
+    # the proxy takes a shallow snapshot of the body before this hook runs
+    body = {"input": "Email Marie Dupont at marie@acme.com"}
+    data = {
+        "input": "Email Marie Dupont at marie@acme.com",
+        "proxy_server_request": {"body": body},
+    }
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+
+    assert "Marie Dupont" not in out["input"]
+    assert "<PERSON_1>" in out["input"]
+    # the string-rebind would otherwise leave raw PII in the snapshot body
+    assert body["input"] == out["input"]
+    assert "Marie Dupont" not in body["input"]
+    assert out["metadata"]["privaite_map"] == _FAKES
+
+
+@pytest.mark.asyncio
+async def test_pre_call_anonymizes_responses_role_message_list_in_place():
+    gr = _make_guardrail()
+    input_value = [{"role": "user", "content": "Hi Marie Dupont"}]
+    data = {"input": input_value}
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+    # mutated in place (same list object)
+    assert out["input"] is input_value
+    assert "<PERSON_1>" in input_value[0]["content"]
+    assert "Marie Dupont" not in input_value[0]["content"]
+
+
+@pytest.mark.asyncio
+async def test_pre_call_anonymizes_responses_content_parts_in_place():
+    gr = _make_guardrail()
+    input_value = [{"type": "input_text", "text": "Hi Marie Dupont"}]
+    data = {"input": input_value}
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+    assert out["input"] is input_value
+    assert input_value[0]["text"] == "Hi <PERSON_1>"
+
+
+@pytest.mark.asyncio
+async def test_pre_call_anonymizes_both_messages_and_input():
+    # a crafted /v1/responses body with decoy messages + PII in input: BOTH must
+    # be anonymized under one shared mapping (input must not be skipped).
+    gr = _make_guardrail()
+    messages = [{"role": "user", "content": "Email Marie Dupont"}]
+    data = {"messages": messages, "input": "also reach marie@acme.com"}
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+
+    assert "Marie Dupont" not in messages[0]["content"]
+    assert "<PERSON_1>" in messages[0]["content"]
+    assert "marie@acme.com" not in out["input"]
+    assert "<EMAIL_ADDRESS_1>" in out["input"]
+    # one consistent map covering both sources
+    assert out["metadata"]["privaite_map"] == _FAKES
+
+
+class _EngineReturningOneMessageFewer(_FakeEngine):
+    async def process_request(self, messages):
+        anonymized, mapping = await super().process_request(messages)
+        return anonymized[:-1], mapping
+
+
+@pytest.mark.asyncio
+async def test_pre_call_refuses_a_request_the_engine_did_not_fully_return(monkeypatch):
+    monkeypatch.setattr(sys.modules["privaite.pii.engine"], "PIIEngine", _EngineReturningOneMessageFewer)
+    gr = _make_guardrail()
+    data = {"messages": [{"role": "user", "content": "Email Marie Dupont"}], "input": "also reach marie@acme.com"}
+
+    messages = data["messages"]
+    with pytest.raises(RuntimeError, match="returned 1 messages for 2"):
+        await gr.async_pre_call_hook(None, None, data, "aresponses")
+    assert data["messages"] is messages and messages[0]["content"] == "Email Marie Dupont", "nothing is half written"
+
+
+@pytest.mark.asyncio
+async def test_pre_call_unhandled_input_shape_is_passthrough():
+    # an unexpected input shape (neither str nor list) is left untouched.
+    gr = _make_guardrail()
+    data = {"input": {"unexpected": "shape"}}
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+    assert "privaite_map" not in out.get("metadata", {})
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_mixed_responses_input_list():
+    # an agentic turn: a role message + a function_call_output + a bare string.
+    # the old homogeneity check wrapped the whole list as one content and left
+    # the non-message items raw; every item must be scanned.
+    gr = _make_guardrail()
+    data = {
+        "input": [
+            {"role": "user", "content": "I am Marie Dupont"},
+            {"type": "function_call_output", "call_id": "c1", "output": "reach marie@acme.com"},
+            "also Marie Dupont",
+        ]
+    }
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+    serialized = str(out["input"])
+
+    assert "Marie Dupont" not in serialized
+    assert "marie@acme.com" not in serialized
+    assert out["input"][1]["type"] == "function_call_output"  # structure kept
+    assert out["input"][1]["call_id"] == "c1"
+    assert out["metadata"]["privaite_map"]
+
+
+@pytest.mark.asyncio
+async def test_pre_call_block_gate_fires_on_mixed_responses_input():
+    # a blocked type inside a function_call_output must reject the request; the
+    # gate is only reachable because the item is scanned in the first place.
+    from fastapi import HTTPException
+
+    gr = _make_guardrail(block_entities=["EMAIL_ADDRESS"])
+    data = {"input": [{"type": "function_call_output", "output": "reach marie@acme.com"}]}
+    with pytest.raises(HTTPException) as ei:
+        await gr.async_pre_call_hook(None, None, data, "aresponses")
+    assert ei.value.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_post_call_restores_responses_output_dict_and_object():
+    gr = _make_guardrail()
+    data = {"metadata": {"privaite_map": _FAKES}}
+
+    # output item as a dict (message with output_text content)
+    dict_item = {
+        "type": "message",
+        "content": [{"type": "output_text", "text": "Hello <PERSON_1>"}],
+    }
+    # output item as an object (function_call with arguments)
+    obj_item = types.SimpleNamespace(
+        type="function_call",
+        content=None,
+        arguments='{"to": "<EMAIL_ADDRESS_1>"}',
+    )
+    response = types.SimpleNamespace(output=[dict_item, obj_item])
+
+    out = await gr.async_post_call_success_hook(data, None, response)
+
+    assert out.output[0]["content"][0]["text"] == "Hello Marie Dupont"
+    assert out.output[1].arguments == '{"to": "marie@acme.com"}'
+    assert "privaite_map" not in data["metadata"]
+
+
+@pytest.mark.asyncio
+async def test_post_call_failure_hook_pops_map():
+    # the success/streaming hooks never run on failure; the failure hook must
+    # still drop the reversible map so it cannot reach a failure spend-log.
+    gr = _make_guardrail()
+    rd = {"metadata": {"privaite_map": dict(_FAKES), "other": 1}}
+    assert await gr.async_post_call_failure_hook(rd, RuntimeError("x"), None) is None
+    assert "privaite_map" not in rd["metadata"]
+    assert rd["metadata"]["other"] == 1
+    # missing metadata and a non-dict request_data are safe no-ops
+    assert await gr.async_post_call_failure_hook({}, RuntimeError("x"), None) is None
+    assert await gr.async_post_call_failure_hook(None, RuntimeError("x"), None) is None
+
+
+def test_block_entities_config_model_field():
+    from litellm.types.proxy.guardrails.guardrail_hooks.privaite import (
+        PrivaiteGuardrailConfigModel,
+    )
+
+    assert PrivaiteGuardrailConfigModel.model_fields["block_entities"].default is None
+
+
+def test_block_entities_parsed_from_string_and_list():
+    assert _make_guardrail(block_entities="US_SSN, CREDIT_CARD").block_entities == ("US_SSN", "CREDIT_CARD")
+    assert _make_guardrail(block_entities=["EMAIL_ADDRESS"]).block_entities == ("EMAIL_ADDRESS",)
+    assert _make_guardrail().block_entities == ()
+
+
+def test_initialize_guardrail_passes_block_entities():
+    from litellm.proxy.guardrails.guardrail_hooks.privaite import initialize_guardrail
+    from litellm.types.guardrails import LitellmParams
+
+    litellm_params = LitellmParams(
+        guardrail="privaite",
+        mode="pre_call",
+        default_on=False,
+        preset="light",
+        languages="en",
+        deanonymize=True,
+        block_entities=["US_SSN"],
+    )
+    callback = initialize_guardrail(litellm_params, {"guardrail_name": "privaite"})
+    assert callback.block_entities == ("US_SSN",)
+
+
+@pytest.mark.asyncio
+async def test_block_entities_rejects_with_400():
+    from fastapi import HTTPException
+
+    gr = _make_guardrail(block_entities=["EMAIL_ADDRESS"])
+    data = {"messages": [{"role": "user", "content": "reach marie@acme.com"}]}
+    with pytest.raises(HTTPException) as ei:
+        await gr.async_pre_call_hook(None, None, data, "completion")
+
+    assert ei.value.status_code == 400
+    detail = ei.value.detail
+    msg = detail["error"] if isinstance(detail, dict) else str(detail)
+    assert "EMAIL_ADDRESS" in msg
+    assert "marie@acme.com" not in msg  # the value never leaks into the error
+    assert "privaite_map" not in (data.get("metadata") or {})  # nothing forwarded
+
+
+@pytest.mark.asyncio
+async def test_block_entities_ignores_types_not_present():
+    # a blocked type that is absent must not disturb a normal request: the other
+    # PII is still masked and the request goes through with a restore map.
+    gr = _make_guardrail(block_entities=["US_SSN"])
+    data = {"messages": [{"role": "user", "content": "I am Marie Dupont, marie@acme.com"}]}
+    out = await gr.async_pre_call_hook(None, None, data, "completion")
+
+    serialized = str(out["messages"])
+    assert "Marie Dupont" not in serialized
+    assert "marie@acme.com" not in serialized
+    assert out.get("metadata", {}).get("privaite_map")
+
+
+@pytest.mark.asyncio
+async def test_block_entities_fails_closed_when_privaite_too_old(monkeypatch):
+    # simulate an older privaite whose PIIConfig has no block_entities field:
+    # extra="allow" would swallow it silently, so the guardrail must refuse.
+    schema = sys.modules["privaite.config.schema"]
+    monkeypatch.setattr(schema.PIIConfig, "model_fields", {"preset": None, "languages": None})
+
+    gr = _make_guardrail(block_entities=["EMAIL_ADDRESS"])
+    data = {"messages": [{"role": "user", "content": "marie@acme.com"}]}
+    with pytest.raises(RuntimeError, match="block_entities"):
+        await gr.async_pre_call_hook(None, None, data, "completion")
+
+
+# --- Responses API: tool output list-of-parts, typed carriers, prompt variables ---
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_custom_tool_call_output_list_of_parts():
+    # a shell/custom tool that read a file returns its bytes as a list of
+    # {type, text} parts (not a bare string); every text leaf must be scrubbed
+    # while the binary image part is relayed whole (base64 carries nothing a
+    # text detector can find, and rewriting it would corrupt the payload).
+    gr = _make_guardrail()
+    output = [
+        {"type": "output_text", "text": "the file says reach marie@acme.com"},
+        {"type": "input_image", "image_url": "data:image/png;base64,QUJD"},
+    ]
+    data = {"input": [{"type": "custom_tool_call_output", "call_id": "c1", "output": output}]}
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+
+    scrubbed = out["input"][0]["output"]
+    assert scrubbed[0]["text"] == "the file says reach <EMAIL_ADDRESS_1>"
+    assert scrubbed[1]["image_url"] == "data:image/png;base64,QUJD"  # binary relayed whole
+    assert out["input"][0]["call_id"] == "c1"
+    assert out["metadata"]["privaite_map"]
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_typed_action_carrier():
+    # a file_search_call carries user data in its `queries`/`results` fields, not
+    # in `content`/`output`; the old field scan never looked there.
+    gr = _make_guardrail()
+    data = {
+        "input": [
+            {
+                "type": "file_search_call",
+                "id": "fs1",
+                "queries": ["records for Marie Dupont", "orders by marie@acme.com"],
+                "results": [{"text": "found Marie Dupont", "score": 0.9}],
+            }
+        ]
+    }
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+    item = out["input"][0]
+    assert item["queries"] == ["records for <PERSON_1>", "orders by <EMAIL_ADDRESS_1>"]
+    assert item["results"][0]["text"] == "found <PERSON_1>"
+    assert item["results"][0]["score"] == 0.9  # non-text leaf untouched
+    assert item["id"] == "fs1"
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_prompt_variables():
+    # Responses prompt-template variables carry user data; the template id/version
+    # do not. A variable can be a bare string or a typed content part.
+    gr = _make_guardrail()
+    data = {
+        "prompt": {
+            "id": "pmpt_123",
+            "version": "2",
+            "variables": {
+                "customer": "Marie Dupont",
+                "note": {"type": "input_text", "text": "email marie@acme.com"},
+            },
+        }
+    }
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+    variables = out["prompt"]["variables"]
+    assert variables["customer"] == "<PERSON_1>"
+    assert variables["note"]["text"] == "email <EMAIL_ADDRESS_1>"
+    assert out["prompt"]["id"] == "pmpt_123"  # template id is not user data
+    assert out["prompt"]["version"] == "2"
+    assert out["metadata"]["privaite_map"]
+
+
+@pytest.mark.asyncio
+async def test_pre_call_block_gate_fires_on_prompt_variables():
+    # the block gate must reach a blocked type sitting only in prompt.variables;
+    # it is only reachable because the variable is scanned in the first place.
+    from fastapi import HTTPException
+
+    gr = _make_guardrail(block_entities=["EMAIL_ADDRESS"])
+    data = {"prompt": {"id": "pmpt_1", "variables": {"to": "marie@acme.com"}}}
+    with pytest.raises(HTTPException) as ei:
+        await gr.async_pre_call_hook(None, None, data, "aresponses")
+    assert ei.value.status_code == 400
+
+
+# --- Auxiliary request fields: completions prompt/suffix, prediction, user_location ---
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_completions_prompt_string_and_suffix_and_fixes_snapshot():
+    gr = _make_guardrail()
+    body = {"prompt": "Complete for Marie Dupont", "suffix": "signed marie@acme.com"}
+    data = {
+        "prompt": "Complete for Marie Dupont",
+        "suffix": "signed marie@acme.com",
+        "proxy_server_request": {"body": body},
+    }
+    out = await gr.async_pre_call_hook(None, None, data, "text_completion")
+
+    assert out["prompt"] == "Complete for <PERSON_1>"
+    assert out["suffix"] == "signed <EMAIL_ADDRESS_1>"
+    # the detached snapshot copy must be overwritten too, or raw PII leaks through it
+    assert body["prompt"] == out["prompt"]
+    assert body["suffix"] == out["suffix"]
+    assert "Marie Dupont" not in body["prompt"]
+    assert out["metadata"]["privaite_map"]
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_completions_prompt_batch_list_and_skips_tokens():
+    # the /v1/completions batch shape: string leaves are scrubbed; a tokenized
+    # (integer-array) prompt passes through unscanned as documented.
+    gr = _make_guardrail()
+    prompt = ["reach marie@acme.com", [1, 15, 27]]
+    data = {"prompt": prompt}
+    out = await gr.async_pre_call_hook(None, None, data, "text_completion")
+    assert out["prompt"][0] == "reach <EMAIL_ADDRESS_1>"
+    assert out["prompt"][1] == [1, 15, 27]  # token array untouched
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_prediction_content():
+    # chat predicted outputs carry the client's current document verbatim.
+    gr = _make_guardrail()
+    prediction = {"type": "content", "content": "draft addressed to Marie Dupont"}
+    data = {"messages": [{"role": "user", "content": "edit this"}], "prediction": prediction}
+    out = await gr.async_pre_call_hook(None, None, data, "completion")
+    # prediction dict is aliased by the body snapshot, so it is rewritten in place
+    assert prediction["content"] == "draft addressed to <PERSON_1>"
+    assert out["metadata"]["privaite_map"]
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_web_search_user_location():
+    gr = _make_guardrail()
+    web_search = {"user_location": {"type": "approximate", "approximate": {"city": "Marie Dupont"}}}
+    data = {"messages": [{"role": "user", "content": "weather?"}], "web_search_options": web_search}
+    await gr.async_pre_call_hook(None, None, data, "completion")
+    assert web_search["user_location"]["approximate"]["city"] == "<PERSON_1>"
+
+
+@pytest.mark.asyncio
+async def test_pre_call_only_aux_field_is_not_skipped():
+    # a request whose ONLY user text is an auxiliary field (no messages/input)
+    # must not be short-circuited by the pre-call early return.
+    gr = _make_guardrail()
+    data = {"suffix": "signed marie@acme.com"}
+    out = await gr.async_pre_call_hook(None, None, data, "text_completion")
+    assert out["suffix"] == "signed <EMAIL_ADDRESS_1>"
+    assert out["metadata"]["privaite_map"]
+
+
+# --- Restore parity: refusal and audio transcript (non-streaming + streaming) ---
+
+
+@pytest.mark.asyncio
+async def test_post_call_restores_refusal_and_audio_transcript():
+    gr = _make_guardrail()
+    data = {"metadata": {"privaite_map": _FAKES}}
+
+    # audio as an object (transcript attribute) and refusal quoting the request.
+    obj_msg = types.SimpleNamespace(
+        content=None,
+        tool_calls=None,
+        function_call=None,
+        refusal="I won't email <PERSON_1>",
+        audio=types.SimpleNamespace(transcript="calling <PERSON_1> now"),
+    )
+    # audio as a dict on a second choice.
+    dict_msg = types.SimpleNamespace(
+        content=None,
+        tool_calls=None,
+        function_call=None,
+        audio={"transcript": "reach <EMAIL_ADDRESS_1>"},
+    )
+    response = types.SimpleNamespace(
+        choices=[types.SimpleNamespace(message=obj_msg), types.SimpleNamespace(message=dict_msg)]
+    )
+    out = await gr.async_post_call_success_hook(data, None, response)
+
+    assert out.choices[0].message.refusal == "I won't email Marie Dupont"
+    assert out.choices[0].message.audio.transcript == "calling Marie Dupont now"
+    assert out.choices[1].message.audio["transcript"] == "reach marie@acme.com"
+
+
+@pytest.mark.asyncio
+async def test_streaming_restores_refusal():
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    def _chunk(refusal, finish=None):
+        return types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    index=0,
+                    delta=types.SimpleNamespace(content=None, tool_calls=None, function_call=None, refusal=refusal),
+                    finish_reason=finish,
+                )
+            ]
+        )
+
+    async def _source():
+        yield _chunk("I won't reach <PER")  # placeholder split across chunks
+        yield _chunk("SON_1>", finish="stop")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    refusal = "".join(getattr(c.choices[0].delta, "refusal", None) or "" for c in chunks)
+    assert refusal == "I won't reach Marie Dupont"
+
+
+@pytest.mark.asyncio
+async def test_streaming_restores_audio_transcript():
+    # streamed audio transcript fragments are de-anonymized with a placeholder
+    # split across chunks reassembled in the audio-segment buffer.
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    def _chunk(transcript, finish=None):
+        return types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    index=0,
+                    delta=types.SimpleNamespace(
+                        content=None,
+                        tool_calls=None,
+                        function_call=None,
+                        audio=types.SimpleNamespace(transcript=transcript),
+                    ),
+                    finish_reason=finish,
+                )
+            ]
+        )
+
+    async def _source():
+        yield _chunk("Hi <PER")
+        yield _chunk("SON_1>", finish="stop")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    transcript = "".join(c.choices[0].delta.audio.transcript or "" for c in chunks if c.choices[0].delta.audio)
+    assert transcript == "Hi Marie Dupont"
+
+
+def _bare_delta(**fields):
+    """A streamed delta carrying only the given fields (the others None), the
+    shape of the finish chunk most providers send: delta {} + finish_reason."""
+    base = dict(content=None, tool_calls=None, function_call=None)
+    base.update(fields)
+    return types.SimpleNamespace(**base)
+
+
+def _choice_chunk(delta, finish=None, index=0):
+    return types.SimpleNamespace(choices=[types.SimpleNamespace(index=index, delta=delta, finish_reason=finish)])
+
+
+def _get(obj, *path):
+    """Walk dict-or-object attributes; the guardrail may create a carrier as a
+    plain dict on a finish delta that had none."""
+    for key in path:
+        if obj is None:
+            return None
+        obj = obj.get(key) if isinstance(obj, dict) else getattr(obj, key, None)
+    return obj
+
+
+# The restorer holds back a tail that could begin a placeholder ("<PER" here),
+# so a channel ending on one loses it unless the end of the stream flushes it.
+
+
+@pytest.mark.parametrize("field", ["reasoning_content", "reasoning", "refusal"])
+async def test_streaming_flushes_text_field_tail_onto_a_bare_finish_chunk(field):
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    async def _source():
+        yield _choice_chunk(_bare_delta(**{field: "Considering <PERSON_1>, then <PER"}))
+        yield _choice_chunk(_bare_delta(**{field: None}), finish="stop")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    out = "".join(getattr(choice.delta, field, None) or "" for chunk in chunks for choice in chunk.choices)
+    assert out == "Considering Marie Dupont, then <PER"
+
+
+async def test_streaming_flushes_tool_tail_onto_a_bare_finish_chunk():
+    # A provider that stops mid-argument (finish_reason "length") sends the
+    # finish on a chunk with no tool_calls slot. The held fragment must still
+    # go out, on that chunk, under the same tool_call index: the client
+    # reassembles arguments per index and must see every byte it was sent.
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    async def _source():
+        call = types.SimpleNamespace(index=2, function=types.SimpleNamespace(arguments='{"to": "<EMAIL'))
+        yield _choice_chunk(_bare_delta(tool_calls=[call]))
+        yield _choice_chunk(_bare_delta(), finish="length")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    args = {}
+    for chunk in chunks:
+        for choice in chunk.choices:
+            for tc in choice.delta.tool_calls or []:
+                slot = _get(tc, "index")
+                args[slot] = args.get(slot, "") + (_get(tc, "function", "arguments") or "")
+    assert args == {2: '{"to": "<EMAIL'}
+    assert chunks[-1].choices[0].finish_reason == "length"
+    assert chunks[-1].choices[0].delta.tool_calls
+
+
+async def test_streaming_flushes_function_call_tail_onto_a_bare_finish_chunk():
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    async def _source():
+        yield _choice_chunk(_bare_delta(function_call=types.SimpleNamespace(arguments='{"n": "<PER')))
+        yield _choice_chunk(_bare_delta(), finish="length")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    out = "".join(
+        _get(choice.delta, "function_call", "arguments") or "" for chunk in chunks for choice in chunk.choices
+    )
+    assert out == '{"n": "<PER'
+
+
+async def test_streaming_drains_held_tails_when_the_stream_ends_without_finish_reason():
+    # Some providers close the stream without ever sending finish_reason. What
+    # the buffers still hold goes out as trailing chunks instead of vanishing
+    # with the generator: one per held channel, cloned from the last chunk of
+    # its choice, without that chunk's usage and without a finish_reason.
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    async def _source():
+        first = _choice_chunk(_bare_delta(content="Ask <PERSON_1> or <PER", audio={"transcript": "Call <EMAIL"}))
+        first.usage = types.SimpleNamespace(total_tokens=3)
+        yield first
+        call = types.SimpleNamespace(index=0, function=types.SimpleNamespace(arguments='{"to": "<EMAIL'))
+        yield _choice_chunk(_bare_delta(tool_calls=[call]), index=1)
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    content = audio = args = ""
+    for chunk in chunks:
+        for choice in chunk.choices:
+            assert choice.finish_reason is None
+            if choice.index == 0:
+                content += choice.delta.content or ""
+                audio += _get(choice.delta, "audio", "transcript") or ""
+            else:
+                for tc in choice.delta.tool_calls or []:
+                    args += _get(tc, "function", "arguments") or ""
+    assert content == "Ask Marie Dupont or <PER"
+    assert audio == "Call <EMAIL"
+    assert args == '{"to": "<EMAIL'
+    assert chunks[0].usage.total_tokens == 3
+    assert all(getattr(chunk, "usage", None) is None for chunk in chunks[2:])
+    assert len(chunks) == 5  # 2 from the provider + content, audio and tool tails
+
+
+async def test_streaming_finish_chunk_with_empty_slots_gets_the_tails_under_them():
+    # A finish chunk may carry the tool_call slot and the function_call holder
+    # with empty arguments. Their restore skips an empty fragment, so the sweep
+    # appends the held tail under the existing holder: no duplicate slot.
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    def _slots(args, fc_args):
+        return _bare_delta(
+            tool_calls=[types.SimpleNamespace(index=2, function=types.SimpleNamespace(arguments=args))],
+            function_call=types.SimpleNamespace(arguments=fc_args),
+        )
+
+    async def _source():
+        yield _choice_chunk(_slots('{"to": "<EMAIL', '{"n": "<PER'))
+        yield _choice_chunk(_slots("", ""), finish="length")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    finish = chunks[-1].choices[0].delta
+    assert len(finish.tool_calls) == 1
+    assert finish.tool_calls[0].index == 2
+    assert finish.tool_calls[0].function.arguments == "<EMAIL"
+    assert finish.function_call.arguments == "<PER"
+
+
+async def test_streaming_tails_on_real_litellm_chunks_stay_serializable():
+    # The same two shapes on litellm's own chunk types: a carrier created on a
+    # bare finish Delta and a drained clone must both still serialize the way
+    # the proxy does it (model_dump_json).
+    from litellm.types.utils import ChatCompletionDeltaToolCall, Delta, Function, ModelResponseStream, StreamingChoices
+
+    gr = _make_guardrail()
+
+    def _chunk(delta, finish=None):
+        choice = StreamingChoices(index=0, delta=delta, finish_reason=finish)
+        return ModelResponseStream(id="chunk", model="m", choices=[choice])
+
+    async def _source():
+        call = ChatCompletionDeltaToolCall(index=0, type="function", function=Function(arguments='{"to": "<EMAIL'))
+        yield _chunk(Delta(content=None, tool_calls=[call]))
+        yield _chunk(Delta(content=None), finish="length")
+
+    async def _no_finish():
+        yield _chunk(Delta(content=None, reasoning_content="then <PER"))
+
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    payloads = [json.loads(chunk.model_dump_json(exclude_none=True)) for chunk in chunks]
+    finish = payloads[-1]["choices"][0]
+    assert finish["finish_reason"] == "length"
+    assert finish["delta"]["tool_calls"][0]["index"] == 0
+    assert finish["delta"]["tool_calls"][0]["function"]["arguments"] == "<EMAIL"
+
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _no_finish(), request_data))
+    payloads = [json.loads(chunk.model_dump_json(exclude_none=True)) for chunk in chunks]
+    assert [p["choices"][0]["delta"].get("reasoning_content") for p in payloads] == ["then ", "<PER"]
+    assert "finish_reason" not in payloads[-1]["choices"][0]
+    assert payloads[-1]["id"] == "chunk"
+
+
+# An original carrying the characters a JSON string literal must escape.
+_SPECIALS_ORIGINAL = 'Jean "JJ" O\'Neil\\n\tnext'
+
+
+async def test_post_call_restores_tool_arguments_as_valid_json(monkeypatch):
+    # A restored original holding a quote, a backslash or a newline must be
+    # spliced JSON-escaped, on every argument carrier: tool_calls, the legacy
+    # function_call and a Responses function_call output item. A plain
+    # substitution would leave the client's json.loads failing on the quote.
+    monkeypatch.setitem(_FAKES, "<PERSON_2>", _SPECIALS_ORIGINAL)
+    gr = _make_guardrail()
+    data = {"metadata": {"privaite_map": _FAKES}}
+    message = types.SimpleNamespace(
+        content=None,
+        tool_calls=[
+            types.SimpleNamespace(function=types.SimpleNamespace(arguments='{"name": "<PERSON_2>", "n": 1}')),
+            # not JSON: the plain string restore is all there is, byte for byte
+            types.SimpleNamespace(function=types.SimpleNamespace(arguments="name=<PERSON_2>")),
+        ],
+        function_call=types.SimpleNamespace(arguments='["<PERSON_2>", {"k": "<PERSON_1>"}]'),
+    )
+    response = types.SimpleNamespace(
+        choices=[types.SimpleNamespace(message=message)],
+        output=[{"type": "function_call", "arguments": '{"who": "<PERSON_2>"}'}],
+    )
+
+    out = await gr.async_post_call_success_hook(data, None, response)
+
+    msg = out.choices[0].message
+    assert json.loads(msg.tool_calls[0].function.arguments) == {"name": _SPECIALS_ORIGINAL, "n": 1}
+    assert msg.tool_calls[1].function.arguments == f"name={_SPECIALS_ORIGINAL}"
+    assert json.loads(msg.function_call.arguments) == [_SPECIALS_ORIGINAL, {"k": "Marie Dupont"}]
+    assert json.loads(out.output[0]["arguments"]) == {"who": _SPECIALS_ORIGINAL}
+
+
+async def test_post_call_untouched_arguments_are_kept_byte_for_byte():
+    # No placeholder in the arguments: the string is returned as received, not
+    # re-encoded (key order, spacing and escapes stay the provider's).
+    gr = _make_guardrail()
+    data = {"metadata": {"privaite_map": _FAKES}}
+    raw = '{"b":1,\n "a": "x\\u00e9"}'
+    message = types.SimpleNamespace(
+        content=None,
+        tool_calls=[types.SimpleNamespace(function=types.SimpleNamespace(arguments=raw))],
+        function_call=None,
+    )
+    response = types.SimpleNamespace(choices=[types.SimpleNamespace(message=message)])
+    out = await gr.async_post_call_success_hook(data, None, response)
+    assert out.choices[0].message.tool_calls[0].function.arguments == raw
+
+
+async def test_streaming_tool_arguments_stay_valid_json_with_special_chars(monkeypatch):
+    # Streamed argument fragments are JSON source text: the original spliced in
+    # must be its JSON-escaped form, so the reassembled arguments still parse.
+    # Content keeps the raw original. The placeholder is split across chunks.
+    monkeypatch.setitem(_FAKES, "<PERSON_2>", _SPECIALS_ORIGINAL)
+    gr = _make_guardrail()
+    request_data = {"metadata": {"privaite_map": _FAKES}}
+
+    def _delta(args, content=None):
+        call = types.SimpleNamespace(index=0, function=types.SimpleNamespace(arguments=args))
+        return _bare_delta(content=content, tool_calls=[call], function_call=types.SimpleNamespace(arguments=args))
+
+    async def _source():
+        yield _choice_chunk(_delta('{"name": "<PERS', content="Hi <PERS"))
+        yield _choice_chunk(_delta('ON_2>"}', content="ON_2>"), finish="stop")
+
+    chunks = await _collect(gr.async_post_call_streaming_iterator_hook(None, _source(), request_data))
+    tool_args = "".join(
+        tc.function.arguments for chunk in chunks for choice in chunk.choices for tc in (choice.delta.tool_calls or [])
+    )
+    fc_args = "".join(choice.delta.function_call.arguments for chunk in chunks for choice in chunk.choices)
+    content = "".join(choice.delta.content or "" for chunk in chunks for choice in chunk.choices)
+    assert json.loads(tool_args) == {"name": _SPECIALS_ORIGINAL}
+    assert json.loads(fc_args) == {"name": _SPECIALS_ORIGINAL}
+    assert content == f"Hi {_SPECIALS_ORIGINAL}"
+
+
+@pytest.mark.asyncio
+async def test_pre_call_scans_a_responses_item_whose_type_is_not_a_string():
+    gr = _make_guardrail()
+    data = {"input": [{"type": 5, "output": "Email Marie Dupont"}]}
+
+    out = await gr.async_pre_call_hook(None, None, data, "aresponses")
+
+    assert out["input"][0]["output"] == "Email <PERSON_1>"
+
+
+@pytest.mark.asyncio
+async def test_pre_call_stashes_the_map_when_metadata_is_null():
+    gr = _make_guardrail()
+    data = {"messages": [{"role": "user", "content": "Email Marie Dupont"}], "metadata": None}
+
+    out = await gr.async_pre_call_hook(None, None, data, "acompletion")
+
+    assert out["metadata"]["privaite_map"] == {"<PERSON_1>": "Marie Dupont"}
+
+
+@pytest.mark.asyncio
+async def test_post_call_restores_a_dict_shaped_response():
+    gr = _make_guardrail()
+    response = {"choices": [{"message": {"content": "Sending to <PERSON_1>"}}]}
+
+    out = await gr.async_post_call_success_hook({"metadata": {"privaite_map": _FAKES}}, None, response)
+
+    assert out["choices"][0]["message"]["content"] == "Sending to Marie Dupont"
+
+
+@pytest.mark.asyncio
+async def test_engine_for_a_privaite_without_pii_engine_recommends_the_install(monkeypatch):
+    monkeypatch.delattr(sys.modules["privaite.pii.engine"], "PIIEngine")
+
+    with pytest.raises(RuntimeError, match=r"pip install 'privaite>=0\.7\.2'"):
+        await _make_guardrail()._engine_for(["en"])
+
+
+@pytest.mark.asyncio
+async def test_post_call_restores_a_copy_and_leaves_the_response_litellm_caches_and_logs():
+    gr = _make_guardrail()
+    message = types.SimpleNamespace(content="Sending to <PERSON_1>", tool_calls=None, function_call=None)
+    response = types.SimpleNamespace(choices=[types.SimpleNamespace(message=message)])
+
+    out = await gr.async_post_call_success_hook({"metadata": {"privaite_map": _FAKES}}, None, response)
+
+    assert out.choices[0].message.content == "Sending to Marie Dupont"
+    assert message.content == "Sending to <PERSON_1>", "the object LiteLLM caches and logs must keep the placeholder"
+
+
+@pytest.mark.asyncio
+async def test_streaming_restores_copies_and_leaves_the_chunks_litellm_reassembles():
+    gr = _make_guardrail()
+    source = [
+        _choice_chunk(_bare_delta(content="Hi <PERS")),
+        _choice_chunk(_bare_delta(content="ON_1>"), finish="stop"),
+    ]
+
+    async def _source():
+        for chunk in source:
+            yield chunk
+
+    chunks = await _collect(
+        gr.async_post_call_streaming_iterator_hook(None, _source(), {"metadata": {"privaite_map": _FAKES}})
+    )
+
+    assert "".join(choice.delta.content or "" for chunk in chunks for choice in chunk.choices) == "Hi Marie Dupont"
+    assert [chunk.choices[0].delta.content for chunk in source] == ["Hi <PERS", "ON_1>"], (
+        "the chunks LiteLLM keeps to rebuild the logged and cached response must keep the placeholder"
+    )