diff --git a/litellm/litellm_core_utils/redact_messages.py b/litellm/litellm_core_utils/redact_messages.py
index f3f560b33b..0d7a204c4e 100644
--- a/litellm/litellm_core_utils/redact_messages.py
+++ b/litellm/litellm_core_utils/redact_messages.py
@@ -60,6 +60,9 @@ def _redact_choice_content(choice):
 def _redact_responses_api_output(output_items):
     """Helper to redact ResponsesAPIResponse output items."""
     for output_item in output_items:
+        if hasattr(output_item, "text"):
+            output_item.text = "redacted-by-litellm"
+
         if hasattr(output_item, "content") and isinstance(output_item.content, list):
             for content_part in output_item.content:
                 if hasattr(content_part, "text"):
@@ -75,6 +78,28 @@ def _redact_responses_api_output(output_items):
                         summary_item.text = "redacted-by-litellm"
 
 
+def _redact_responses_api_output_dict(output_items, redacted_str: str):
+    """Helper to redact ResponsesAPIResponse output items in dict form."""
+    for output_item in output_items:
+        if not isinstance(output_item, dict):
+            continue
+
+        if "text" in output_item:
+            output_item["text"] = redacted_str
+
+        if isinstance(output_item.get("content"), list):
+            for content_item in output_item["content"]:
+                if isinstance(content_item, dict) and "text" in content_item:
+                    content_item["text"] = redacted_str
+
+        if output_item.get("type") == "reasoning" and isinstance(
+            output_item.get("summary"), list
+        ):
+            for summary_item in output_item["summary"]:
+                if isinstance(summary_item, dict) and "text" in summary_item:
+                    summary_item["text"] = redacted_str
+
+
 def _redact_standard_logging_object(model_call_details: dict):
     """Redact messages and response inside standard_logging_object if present."""
     standard_logging_object = model_call_details.get("standard_logging_object")
@@ -93,15 +118,7 @@ def _redact_standard_logging_object(model_call_details: dict):
         if isinstance(response, dict) and "output" in response:
             # ResponsesAPIResponse format - redact content in output items
             if isinstance(response.get("output"), list):
-                for output_item in response["output"]:
-                    if isinstance(output_item, dict) and "content" in output_item:
-                        if isinstance(output_item["content"], list):
-                            for content_item in output_item["content"]:
-                                if (
-                                    isinstance(content_item, dict)
-                                    and "text" in content_item
-                                ):
-                                    content_item["text"] = redacted_str
+                _redact_responses_api_output_dict(response["output"], redacted_str)
         elif isinstance(response, dict) and "choices" in response:
             # ModelResponse dict format - redact content in choices
             if isinstance(response.get("choices"), list):
@@ -122,6 +139,29 @@ def _redact_standard_logging_object(model_call_details: dict):
             standard_logging_object["response"] = {"text": redacted_str}
 
 
+def _redact_model_response_dict_choices(choices, redacted_str: str):
+    for choice in choices:
+        if isinstance(choice, dict):
+            if "message" in choice and isinstance(choice["message"], dict):
+                choice["message"]["content"] = redacted_str
+                if "reasoning_content" in choice["message"]:
+                    choice["message"]["reasoning_content"] = redacted_str
+                if "thinking_blocks" in choice["message"]:
+                    choice["message"]["thinking_blocks"] = None
+                if "audio" in choice["message"]:
+                    choice["message"]["audio"] = None
+            elif "delta" in choice and isinstance(choice["delta"], dict):
+                choice["delta"]["content"] = redacted_str
+                if "reasoning_content" in choice["delta"]:
+                    choice["delta"]["reasoning_content"] = redacted_str
+                if "thinking_blocks" in choice["delta"]:
+                    choice["delta"]["thinking_blocks"] = None
+                if "audio" in choice["delta"]:
+                    choice["delta"]["audio"] = None
+        else:
+            _redact_choice_content(choice)
+
+
 def perform_redaction(model_call_details: dict, result):
     """
     Performs the actual redaction on the logging object and result.
@@ -132,6 +172,7 @@ def perform_redaction(model_call_details: dict, result):
     ]
     model_call_details["prompt"] = ""
     model_call_details["input"] = ""
+    _redact_standard_logging_object(model_call_details)
 
     # Redact streaming response
     if (
@@ -171,30 +212,14 @@ def perform_redaction(model_call_details: dict, result):
         elif isinstance(_result, dict) and "choices" in _result:
             # Handle dict representation of ModelResponse (e.g., from model_dump())
             if _result.get("choices") is not None:
-                for choice in _result["choices"]:
-                    if isinstance(choice, dict):
-                        if "message" in choice and isinstance(choice["message"], dict):
-                            choice["message"]["content"] = "redacted-by-litellm"
-                            if "reasoning_content" in choice["message"]:
-                                choice["message"][
-                                    "reasoning_content"
-                                ] = "redacted-by-litellm"
-                            if "thinking_blocks" in choice["message"]:
-                                choice["message"]["thinking_blocks"] = None
-                            if "audio" in choice["message"]:
-                                choice["message"]["audio"] = None
-                        elif "delta" in choice and isinstance(choice["delta"], dict):
-                            choice["delta"]["content"] = "redacted-by-litellm"
-                            if "reasoning_content" in choice["delta"]:
-                                choice["delta"][
-                                    "reasoning_content"
-                                ] = "redacted-by-litellm"
-                            if "thinking_blocks" in choice["delta"]:
-                                choice["delta"]["thinking_blocks"] = None
-                            if "audio" in choice["delta"]:
-                                choice["delta"]["audio"] = None
-                    else:
-                        _redact_choice_content(choice)
+                _redact_model_response_dict_choices(
+                    _result["choices"], "redacted-by-litellm"
+                )
+        elif isinstance(_result, dict) and "output" in _result:
+            if isinstance(_result.get("output"), list):
+                _redact_responses_api_output_dict(
+                    _result["output"], "redacted-by-litellm"
+                )
         elif isinstance(_result, litellm.ResponsesAPIResponse):
             if hasattr(_result, "output"):
                 _redact_responses_api_output(_result.output)
@@ -214,12 +239,15 @@ def should_redact_message_logging(model_call_details: dict) -> bool:
     Determine if message logging should be redacted.
 
     Priority order:
-    1. Dynamic parameter (turn_off_message_logging in request)
-    2. Headers (litellm-disable-message-redaction / litellm-enable-message-redaction)
-    3. Global setting (litellm.turn_off_message_logging)
+    1. Global setting (litellm.turn_off_message_logging)
+    2. Dynamic parameter (turn_off_message_logging in request)
+    3. Headers (litellm-disable-message-redaction / litellm-enable-message-redaction)
     """
     litellm_params = model_call_details.get("litellm_params", {})
 
+    if litellm.turn_off_message_logging is True:
+        return True
+
     metadata_field = get_metadata_variable_name_from_kwargs(litellm_params)
     metadata = litellm_params.get(metadata_field, {})
     if not isinstance(metadata, dict):
@@ -231,13 +259,6 @@ def should_redact_message_logging(model_call_details: dict) -> bool:
     # Get headers from the metadata
     request_headers = metadata.get("headers", {})
 
-    # Check for headers that explicitly control redaction
-    if request_headers and bool(
-        request_headers.get("litellm-disable-message-redaction", False)
-    ):
-        # User explicitly disabled redaction via header
-        return False
-
     possible_enable_headers = [
         "litellm-enable-message-redaction",  # old header. maintain backwards compatibility
         "x-litellm-enable-message-redaction",  # new header
@@ -257,12 +278,18 @@ def should_redact_message_logging(model_call_details: dict) -> bool:
         # Dynamic parameter is explicitly set, use it
         return dynamic_turn_off
 
+    if request_headers and bool(
+        request_headers.get("litellm-disable-message-redaction", False)
+    ):
+        # User explicitly disabled redaction via header
+        return False
+
     # Priority 2: Check if header explicitly enables redaction
     if is_redaction_enabled_via_header:
         return True
 
     # Priority 3: Fall back to global setting
-    return litellm.turn_off_message_logging is True
+    return False
 
 
 def redact_message_input_output_from_logging(
diff --git a/litellm/proxy/common_utils/callback_utils.py b/litellm/proxy/common_utils/callback_utils.py
index 7ddd722a80..198d9503cb 100644
--- a/litellm/proxy/common_utils/callback_utils.py
+++ b/litellm/proxy/common_utils/callback_utils.py
@@ -14,6 +14,8 @@ from litellm.types.utils import (
 blue_color_code = "\033[94m"
 reset_color_code = "\033[0m"
 
+TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY = "_pillar_response_headers_trusted"
+
 if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
 
@@ -417,10 +419,19 @@ def get_logging_caching_headers(request_data: Dict) -> Optional[Dict]:
     if "semantic-similarity" in _metadata:
         headers["x-litellm-semantic-similarity"] = str(_metadata["semantic-similarity"])
 
+    is_trusted_pillar_metadata = (
+        _metadata.get(TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY) is True
+    )
     pillar_headers = _metadata.get("pillar_response_headers")
-    if isinstance(pillar_headers, dict):
-        headers.update(pillar_headers)
-    elif "pillar_flagged" in _metadata:
+    if is_trusted_pillar_metadata and isinstance(pillar_headers, dict):
+        headers.update(
+            {
+                key: str(value)
+                for key, value in pillar_headers.items()
+                if isinstance(key, str) and key.lower().startswith("x-pillar-")
+            }
+        )
+    elif is_trusted_pillar_metadata and "pillar_flagged" in _metadata:
         headers["x-pillar-flagged"] = str(_metadata["pillar_flagged"]).lower()
 
     return headers
diff --git a/litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py b/litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py
index e5df26b7c2..bb1db3d62d 100644
--- a/litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py
+++ b/litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py
@@ -71,6 +71,7 @@ from litellm.types.utils import (
 )
 
 GUARDRAIL_NAME = "bedrock"
+_BEDROCK_DYNAMIC_BODY_DENYLIST = frozenset({"content", "source"})
 
 
 class GuardrailMessageFilterResult(NamedTuple):
@@ -413,11 +414,18 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
         )
         api_key: Optional[str] = None
         if request_data:
-            bedrock_request_data.update(
+            dynamic_request_body_params = (
                 self.get_guardrail_dynamic_request_body_params(
                     request_data=request_data
                 )
             )
+            bedrock_request_data.update(
+                {
+                    key: value
+                    for key, value in dynamic_request_body_params.items()
+                    if key not in _BEDROCK_DYNAMIC_BODY_DENYLIST
+                }
+            )
             if request_data.get("api_key") is not None:
                 api_key = request_data["api_key"]
 
diff --git a/litellm/proxy/guardrails/guardrail_hooks/pillar/pillar.py b/litellm/proxy/guardrails/guardrail_hooks/pillar/pillar.py
index 1b3f11e56f..c9c73053a0 100644
--- a/litellm/proxy/guardrails/guardrail_hooks/pillar/pillar.py
+++ b/litellm/proxy/guardrails/guardrail_hooks/pillar/pillar.py
@@ -29,6 +29,7 @@ from litellm.llms.custom_httpx.http_handler import (
 )
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.common_utils.callback_utils import (
+    TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY,
     add_guardrail_to_applied_guardrails_header,
     get_metadata_variable_name_from_kwargs,
 )
@@ -144,6 +145,7 @@ def build_pillar_response_headers(metadata_store: Dict[str, Any]) -> Dict[str, s
 
     if headers:
         metadata_store["pillar_response_headers"] = headers
+        metadata_store[TRUSTED_PILLAR_RESPONSE_HEADERS_METADATA_KEY] = True
 
     return headers
 
diff --git a/litellm/proxy/hooks/key_management_event_hooks.py b/litellm/proxy/hooks/key_management_event_hooks.py
index 5cdd9ddb4b..6bb9c1f507 100644
--- a/litellm/proxy/hooks/key_management_event_hooks.py
+++ b/litellm/proxy/hooks/key_management_event_hooks.py
@@ -41,6 +41,7 @@ class KeyManagementEventHooks:
         """
         from litellm.proxy.management_helpers.audit_logs import (
             create_audit_log_for_update,
+            get_audit_log_changed_by,
         )
         from litellm.proxy.proxy_server import litellm_proxy_admin_name
 
@@ -61,9 +62,11 @@ class KeyManagementEventHooks:
                     request_data=LiteLLM_AuditLogs(
                         id=str(uuid.uuid4()),
                         updated_at=datetime.now(timezone.utc),
-                        changed_by=litellm_changed_by
-                        or user_api_key_dict.user_id
-                        or litellm_proxy_admin_name,
+                        changed_by=get_audit_log_changed_by(
+                            litellm_changed_by=litellm_changed_by,
+                            user_api_key_dict=user_api_key_dict,
+                            litellm_proxy_admin_name=litellm_proxy_admin_name,
+                        ),
                         changed_by_api_key=user_api_key_dict.api_key,
                         table_name=LitellmTableNames.KEY_TABLE_NAME,
                         object_id=response.token_id or "",
@@ -102,6 +105,7 @@ class KeyManagementEventHooks:
         """
         from litellm.proxy.management_helpers.audit_logs import (
             create_audit_log_for_update,
+            get_audit_log_changed_by,
         )
         from litellm.proxy.proxy_server import litellm_proxy_admin_name
 
@@ -117,9 +121,11 @@ class KeyManagementEventHooks:
                     request_data=LiteLLM_AuditLogs(
                         id=str(uuid.uuid4()),
                         updated_at=datetime.now(timezone.utc),
-                        changed_by=litellm_changed_by
-                        or user_api_key_dict.user_id
-                        or litellm_proxy_admin_name,
+                        changed_by=get_audit_log_changed_by(
+                            litellm_changed_by=litellm_changed_by,
+                            user_api_key_dict=user_api_key_dict,
+                            litellm_proxy_admin_name=litellm_proxy_admin_name,
+                        ),
                         changed_by_api_key=user_api_key_dict.api_key,
                         table_name=LitellmTableNames.KEY_TABLE_NAME,
                         object_id=data.key,
@@ -140,6 +146,7 @@ class KeyManagementEventHooks:
     ):
         from litellm.proxy.management_helpers.audit_logs import (
             create_audit_log_for_update,
+            get_audit_log_changed_by,
         )
         from litellm.proxy.proxy_server import litellm_proxy_admin_name
 
@@ -189,9 +196,11 @@ class KeyManagementEventHooks:
                     request_data=LiteLLM_AuditLogs(
                         id=str(uuid.uuid4()),
                         updated_at=datetime.now(timezone.utc),
-                        changed_by=litellm_changed_by
-                        or user_api_key_dict.user_id
-                        or litellm_proxy_admin_name,
+                        changed_by=get_audit_log_changed_by(
+                            litellm_changed_by=litellm_changed_by,
+                            user_api_key_dict=user_api_key_dict,
+                            litellm_proxy_admin_name=litellm_proxy_admin_name,
+                        ),
                         changed_by_api_key=user_api_key_dict.token,
                         table_name=LitellmTableNames.KEY_TABLE_NAME,
                         object_id=existing_key_row.token,
@@ -220,6 +229,7 @@ class KeyManagementEventHooks:
         """
         from litellm.proxy.management_helpers.audit_logs import (
             create_audit_log_for_update,
+            get_audit_log_changed_by,
         )
         from litellm.proxy.proxy_server import litellm_proxy_admin_name
 
@@ -237,9 +247,11 @@ class KeyManagementEventHooks:
                         request_data=LiteLLM_AuditLogs(
                             id=str(uuid.uuid4()),
                             updated_at=datetime.now(timezone.utc),
-                            changed_by=litellm_changed_by
-                            or user_api_key_dict.user_id
-                            or litellm_proxy_admin_name,
+                            changed_by=get_audit_log_changed_by(
+                                litellm_changed_by=litellm_changed_by,
+                                user_api_key_dict=user_api_key_dict,
+                                litellm_proxy_admin_name=litellm_proxy_admin_name,
+                            ),
                             changed_by_api_key=user_api_key_dict.token,
                             table_name=LitellmTableNames.KEY_TABLE_NAME,
                             object_id=key.token,
diff --git a/litellm/proxy/hooks/user_management_event_hooks.py b/litellm/proxy/hooks/user_management_event_hooks.py
index 38623f9209..08fa8d4dfa 100644
--- a/litellm/proxy/hooks/user_management_event_hooks.py
+++ b/litellm/proxy/hooks/user_management_event_hooks.py
@@ -192,13 +192,19 @@ class UserManagementEventHooks:
         if not litellm.store_audit_logs:
             return
 
+        from litellm.proxy.management_helpers.audit_logs import (
+            get_audit_log_changed_by,
+        )
+
         await create_audit_log_for_update(
             request_data=LiteLLM_AuditLogs(
                 id=str(uuid.uuid4()),
                 updated_at=datetime.now(timezone.utc),
-                changed_by=litellm_changed_by
-                or user_api_key_dict.user_id
-                or litellm_proxy_admin_name,
+                changed_by=get_audit_log_changed_by(
+                    litellm_changed_by=litellm_changed_by,
+                    user_api_key_dict=user_api_key_dict,
+                    litellm_proxy_admin_name=litellm_proxy_admin_name,
+                ),
                 changed_by_api_key=user_api_key_dict.api_key,
                 table_name=LitellmTableNames.USER_TABLE_NAME,
                 object_id=user_id,
diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index 5804e3f8d9..2f7dcb2afb 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -104,6 +104,59 @@ LITELLM_METADATA_ROUTES = (
     "files",
 )
 
+_UNTRUSTED_ROOT_CONTROL_FIELDS = (
+    "proxy_server_request",
+    "standard_logging_object",
+    "secret_fields",
+    "mock_response",
+    "mock_tool_calls",
+    "disable_global_guardrails",
+    "disable_global_guardrail",
+    "opted_out_global_guardrails",
+    "applied_guardrails",
+    "applied_policies",
+    "policy_sources",
+    "pillar_response_headers",
+    "_guardrail_pipelines",
+    "_pipeline_managed_guardrails",
+)
+
+_UNTRUSTED_METADATA_CONTROL_FIELDS = (
+    "disable_global_guardrails",
+    "disable_global_guardrail",
+    "opted_out_global_guardrails",
+    "pillar_response_headers",
+    "_pillar_response_headers_trusted",
+    "pillar_flagged",
+    "pillar_scanners",
+    "pillar_evidence",
+    "pillar_evidence_truncated",
+    "pillar_session_id_response",
+    "applied_guardrails",
+    "applied_policies",
+    "policy_sources",
+    "standard_logging_object",
+    "proxy_server_request",
+    "secret_fields",
+    "_guardrail_pipelines",
+    "_pipeline_managed_guardrails",
+)
+
+_CLIENT_MOCK_CONTROL_FIELDS = frozenset({"mock_response", "mock_tool_calls"})
+_ALLOW_CLIENT_MOCK_RESPONSE_METADATA_KEY = "allow_client_mock_response"
+
+
+def _key_or_team_allows_client_mock_response(
+    user_api_key_dict: UserAPIKeyAuth,
+) -> bool:
+    for admin_metadata in (user_api_key_dict.metadata, user_api_key_dict.team_metadata):
+        if (
+            isinstance(admin_metadata, dict)
+            and admin_metadata.get(_ALLOW_CLIENT_MOCK_RESPONSE_METADATA_KEY) is True
+        ):
+            return True
+    return False
+
 
 def _get_metadata_variable_name(request: Request) -> str:
     """
@@ -962,11 +1015,12 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
     # Strip internal-only keys from user input before the proxy sets its own.
     # These keys are injected by the proxy itself below — user-supplied values
     # must not be trusted.
-    for _internal_key in (
-        "proxy_server_request",
-        "standard_logging_object",
-        "secret_fields",
-    ):
+    _allow_client_mock_response = _key_or_team_allows_client_mock_response(
+        user_api_key_dict
+    )
+    for _internal_key in _UNTRUSTED_ROOT_CONTROL_FIELDS:
+        if _allow_client_mock_response and _internal_key in _CLIENT_MOCK_CONTROL_FIELDS:
+            continue
         data.pop(_internal_key, None)
     # Strip spoofable auth metadata from user-supplied metadata dict
     _user_metadata = data.get("metadata")
@@ -1144,8 +1198,12 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
     for _meta_key in ("metadata", "litellm_metadata"):
         _user_meta = data.get(_meta_key)
         if isinstance(_user_meta, dict):
-            _user_meta.pop("_pipeline_managed_guardrails", None)
-            for _k in [k for k in _user_meta if k.startswith("user_api_key_")]:
+            for _k in [
+                k
+                for k in _user_meta
+                if k.startswith("user_api_key_")
+                or k in _UNTRUSTED_METADATA_CONTROL_FIELDS
+            ]:
                 _user_meta.pop(_k, None)
 
     # Strip caller-supplied routing/budget tags unless the admin has opted
diff --git a/litellm/proxy/management_endpoints/internal_user_endpoints.py b/litellm/proxy/management_endpoints/internal_user_endpoints.py
index c6d37ace4f..66c751f176 100644
--- a/litellm/proxy/management_endpoints/internal_user_endpoints.py
+++ b/litellm/proxy/management_endpoints/internal_user_endpoints.py
@@ -2162,8 +2162,8 @@ async def delete_user(
                     request_data=LiteLLM_AuditLogs(
                         id=str(uuid.uuid4()),
                         updated_at=datetime.now(timezone.utc),
-                        changed_by=litellm_changed_by
-                        or user_api_key_dict.user_id
+                        changed_by=user_api_key_dict.user_id
+                        or litellm_changed_by
                         or litellm_proxy_admin_name,
                         changed_by_api_key=user_api_key_dict.api_key,
                         table_name=LitellmTableNames.USER_TABLE_NAME,
diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py
index 8129fb0de5..6c6a38bcbc 100644
--- a/litellm/proxy/management_endpoints/key_management_endpoints.py
+++ b/litellm/proxy/management_endpoints/key_management_endpoints.py
@@ -5297,8 +5297,8 @@ async def block_key(
                 request_data=LiteLLM_AuditLogs(
                     id=str(uuid.uuid4()),
                     updated_at=datetime.now(timezone.utc),
-                    changed_by=litellm_changed_by
-                    or user_api_key_dict.user_id
+                    changed_by=user_api_key_dict.user_id
+                    or litellm_changed_by
                     or litellm_proxy_admin_name,
                     changed_by_api_key=user_api_key_dict.api_key,
                     table_name=LitellmTableNames.KEY_TABLE_NAME,
@@ -5406,8 +5406,8 @@ async def unblock_key(
                 request_data=LiteLLM_AuditLogs(
                     id=str(uuid.uuid4()),
                     updated_at=datetime.now(timezone.utc),
-                    changed_by=litellm_changed_by
-                    or user_api_key_dict.user_id
+                    changed_by=user_api_key_dict.user_id
+                    or litellm_changed_by
                     or litellm_proxy_admin_name,
                     changed_by_api_key=user_api_key_dict.api_key,
                     table_name=LitellmTableNames.KEY_TABLE_NAME,
@@ -5589,7 +5589,6 @@ async def test_key_logging(
                     "content": "Hello, this is a test from litellm /key/health. No LLM API call was made for this",
                 }
             ],
-            "mock_response": "test response",
         }
         data = await add_litellm_data_to_request(
             data=data,
@@ -5598,6 +5597,7 @@ async def test_key_logging(
             general_settings=general_settings,
             request=request,
         )
+        data["mock_response"] = "test response"
         await litellm.acompletion(
             **data
         )  # make mock completion call to trigger key based callbacks
diff --git a/litellm/proxy/management_endpoints/mcp_management_endpoints.py b/litellm/proxy/management_endpoints/mcp_management_endpoints.py
index fca08e591f..4df01ce109 100644
--- a/litellm/proxy/management_endpoints/mcp_management_endpoints.py
+++ b/litellm/proxy/management_endpoints/mcp_management_endpoints.py
@@ -2230,7 +2230,7 @@ if MCP_AVAILABLE:
                 detail={"error": "Only proxy admins can create MCP toolsets."},
             )
         touched_by = (
-            litellm_changed_by or user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME
+            user_api_key_dict.user_id or litellm_changed_by or LITELLM_PROXY_ADMIN_NAME
         )
         try:
             result = await create_mcp_toolset(prisma_client, payload, touched_by)
@@ -2321,7 +2321,7 @@ if MCP_AVAILABLE:
                 detail={"error": "Only proxy admins can update MCP toolsets."},
             )
         touched_by = (
-            litellm_changed_by or user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME
+            user_api_key_dict.user_id or litellm_changed_by or LITELLM_PROXY_ADMIN_NAME
         )
         try:
             result = await update_mcp_toolset(prisma_client, payload, touched_by)
diff --git a/litellm/proxy/management_endpoints/team_endpoints.py b/litellm/proxy/management_endpoints/team_endpoints.py
index e29b67724c..023e49ba3d 100644
--- a/litellm/proxy/management_endpoints/team_endpoints.py
+++ b/litellm/proxy/management_endpoints/team_endpoints.py
@@ -1174,8 +1174,8 @@ async def new_team(  # noqa: PLR0915
                     request_data=LiteLLM_AuditLogs(
                         id=str(uuid.uuid4()),
                         updated_at=datetime.now(timezone.utc),
-                        changed_by=litellm_changed_by
-                        or user_api_key_dict.user_id
+                        changed_by=user_api_key_dict.user_id
+                        or litellm_changed_by
                         or litellm_proxy_admin_name,
                         changed_by_api_key=user_api_key_dict.api_key,
                         table_name=LitellmTableNames.TEAM_TABLE_NAME,
@@ -1225,8 +1225,8 @@ async def _create_team_update_audit_log(
             request_data=LiteLLM_AuditLogs(
                 id=str(uuid.uuid4()),
                 updated_at=datetime.now(timezone.utc),
-                changed_by=litellm_changed_by
-                or user_api_key_dict.user_id
+                changed_by=user_api_key_dict.user_id
+                or litellm_changed_by
                 or litellm_proxy_admin_name,
                 changed_by_api_key=user_api_key_dict.api_key,
                 table_name=LitellmTableNames.TEAM_TABLE_NAME,
@@ -3054,8 +3054,8 @@ async def delete_team(
                     request_data=LiteLLM_AuditLogs(
                         id=str(uuid.uuid4()),
                         updated_at=datetime.now(timezone.utc),
-                        changed_by=litellm_changed_by
-                        or user_api_key_dict.user_id
+                        changed_by=user_api_key_dict.user_id
+                        or litellm_changed_by
                         or litellm_proxy_admin_name,
                         changed_by_api_key=user_api_key_dict.api_key,
                         table_name=LitellmTableNames.TEAM_TABLE_NAME,
diff --git a/litellm/proxy/management_helpers/audit_logs.py b/litellm/proxy/management_helpers/audit_logs.py
index 7599e11bde..3801a6abd6 100644
--- a/litellm/proxy/management_helpers/audit_logs.py
+++ b/litellm/proxy/management_helpers/audit_logs.py
@@ -23,6 +23,15 @@ from litellm.types.utils import StandardAuditLogPayload
 _audit_log_callback_cache: Dict[str, CustomLogger] = {}
 
 
+def get_audit_log_changed_by(
+    *,
+    litellm_changed_by: Optional[str],
+    user_api_key_dict: UserAPIKeyAuth,
+    litellm_proxy_admin_name: Optional[str],
+) -> Optional[str]:
+    return user_api_key_dict.user_id or litellm_changed_by or litellm_proxy_admin_name
+
+
 def _resolve_audit_log_callback(name: str) -> Optional[CustomLogger]:
     """Resolve a string callback name to a CustomLogger instance, with caching."""
     if name in _audit_log_callback_cache:
@@ -143,8 +152,10 @@ async def create_object_audit_log(
     if _store_audit_logs is not True:
         return
 
-    _changed_by = (
-        litellm_changed_by or user_api_key_dict.user_id or litellm_proxy_admin_name
+    _changed_by = get_audit_log_changed_by(
+        litellm_changed_by=litellm_changed_by,
+        user_api_key_dict=user_api_key_dict,
+        litellm_proxy_admin_name=litellm_proxy_admin_name,
     )
 
     await create_audit_log_for_update(
diff --git a/tests/litellm_utils_tests/test_utils.py b/tests/litellm_utils_tests/test_utils.py
index 20af6e1023..261375aa9a 100644
--- a/tests/litellm_utils_tests/test_utils.py
+++ b/tests/litellm_utils_tests/test_utils.py
@@ -750,7 +750,7 @@ def test_redact_msgs_from_logs_with_dynamic_params():
     In all tests litellm.turn_off_message_logging is True
 
 
-    1. When standard_callback_dynamic_params.turn_off_message_logging is False (or not set): No redaction should occur. User has opted out of redaction.
+    1. When standard_callback_dynamic_params.turn_off_message_logging is False: global redaction still wins.
     2. When standard_callback_dynamic_params.turn_off_message_logging is True: Redaction should occur. User has opted in to redaction.
     3. standard_callback_dynamic_params.turn_off_message_logging not set, litellm.turn_off_message_logging is True: Redaction should occur.
     """
@@ -784,7 +784,7 @@ def test_redact_msgs_from_logs_with_dynamic_params():
         function_id="1234",
     )
 
-    # Test Case 1: standard_callback_dynamic_params = False (or not set)
+    # Test Case 1: standard_callback_dynamic_params = False
     standard_callback_dynamic_params = StandardCallbackDynamicParams(
         turn_off_message_logging=False
     )
@@ -795,8 +795,8 @@ def test_redact_msgs_from_logs_with_dynamic_params():
         result=response_obj,
         model_call_details=litellm_logging_obj.model_call_details,
     )
-    # Assert no redaction occurred
-    assert _redacted_response_obj.choices[0].message.content == test_content
+    # Assert global redaction still occurred
+    assert _redacted_response_obj.choices[0].message.content == "redacted-by-litellm"
 
     # Test Case 2: standard_callback_dynamic_params = True
     standard_callback_dynamic_params = StandardCallbackDynamicParams(
diff --git a/tests/logging_callback_tests/test_logging_redaction_e2e_test.py b/tests/logging_callback_tests/test_logging_redaction_e2e_test.py
index 63ef4bafbb..181f13e06d 100644
--- a/tests/logging_callback_tests/test_logging_redaction_e2e_test.py
+++ b/tests/logging_callback_tests/test_logging_redaction_e2e_test.py
@@ -13,11 +13,13 @@ import logging
 import time
 from unittest.mock import AsyncMock, patch
 
+import httpx
 import pytest
 
 import litellm
 from litellm._logging import verbose_logger
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.responses.main import mock_responses_api_response
 from litellm.types.utils import StandardLoggingPayload
 
 
@@ -126,17 +128,10 @@ async def test_redaction_responses_api():
     test_custom_logger = TestCustomLogger(turn_off_message_logging=True)
     litellm.callbacks = [test_custom_logger]
 
-    # Mock a ResponsesAPIResponse-style response
-    mock_response = {
-        "output": [{"text": "This is a test response"}],
-        "model": "gpt-3.5-turbo",
-        "usage": {"input_tokens": 5, "output_tokens": 5, "total_tokens": 10},
-    }
-
     response = await litellm.aresponses(
         model="gpt-3.5-turbo",
         input="hi",
-        mock_response=mock_response,
+        mock_response="This is a test response",
     )
 
     await asyncio.sleep(1)
@@ -163,6 +158,7 @@ async def test_redaction_responses_api():
                     assert (
                         content_item["text"] == "redacted-by-litellm"
                     ), f"Expected redacted text but got: {content_item['text']}"
+    assert "This is a test response" not in json.dumps(standard_logging_payload)
     print(
         "logged standard logging payload for ResponsesAPIResponse",
         json.dumps(standard_logging_payload, indent=2),
@@ -176,29 +172,36 @@ async def test_redaction_responses_api_stream():
     test_custom_logger = TestCustomLogger(turn_off_message_logging=True)
     litellm.callbacks = [test_custom_logger]
 
-    # Mock a ResponsesAPIResponse-style response with streaming chunks
-    mock_response = [
-        {
-            "output": [{"text": "This"}],
-            "model": "gpt-3.5-turbo",
-        },
-        {
-            "output": [{"text": " is"}],
-            "model": "gpt-3.5-turbo",
-        },
-        {
-            "output": [{"text": " a test response"}],
-            "model": "gpt-3.5-turbo",
-            "usage": {"input_tokens": 5, "output_tokens": 5, "total_tokens": 10},
-        },
-    ]
+    mocked_response_payload = mock_responses_api_response(
+        "This is a test response"
+    ).model_dump()
 
-    response = await litellm.aresponses(
-        model="gpt-3.5-turbo",
-        input="hi",
-        mock_response=mock_response,
-        stream=True,
-    )
+    async def mock_post(self, url, headers, timeout, stream=False, **kwargs):
+        stream_content = (
+            "data: "
+            + json.dumps(
+                {
+                    "type": "response.completed",
+                    "response": mocked_response_payload,
+                }
+            )
+            + "\n\ndata: [DONE]\n\n"
+        )
+        return httpx.Response(
+            status_code=200,
+            content=stream_content,
+            request=httpx.Request("POST", url),
+        )
+
+    with patch(
+        "litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
+        new=mock_post,
+    ):
+        response = await litellm.aresponses(
+            model="gpt-3.5-turbo",
+            input="hi",
+            stream=True,
+        )
 
     # Consume the stream
     chunks = []
@@ -445,18 +448,11 @@ async def test_disable_redaction_header_responses_api():
     test_custom_logger = TestCustomLogger()
     litellm.callbacks = [test_custom_logger]
 
-    # Mock a ResponsesAPIResponse-style response
-    mock_response = {
-        "output": [{"text": "This is a test response"}],
-        "model": "gpt-3.5-turbo",
-        "usage": {"input_tokens": 5, "output_tokens": 5, "total_tokens": 10},
-    }
-
     # Pass the header via litellm_metadata (as the proxy does for Responses API)
     response = await litellm.aresponses(
         model="gpt-3.5-turbo",
         input="hi",
-        mock_response=mock_response,
+        mock_response="This is a test response",
         litellm_metadata={"headers": {"litellm-disable-message-redaction": "true"}},
     )
 
@@ -464,15 +460,16 @@ async def test_disable_redaction_header_responses_api():
     standard_logging_payload = test_custom_logger.logged_standard_logging_payload
     assert standard_logging_payload is not None
 
-    # Verify that messages are NOT redacted because the header was set
+    # Verify that global redaction is not disabled by the header
     print(
         "logged standard logging payload for ResponsesAPI with disable header",
         json.dumps(standard_logging_payload, indent=2, default=str),
     )
 
-    # The content should NOT be redacted
-    assert standard_logging_payload["response"] != {"text": "redacted-by-litellm"}
-    assert standard_logging_payload["messages"][0]["content"] == "hi"
+    response = standard_logging_payload["response"]
+    assert response["output"][0]["content"][0]["text"] == "redacted-by-litellm"
+    assert "This is a test response" not in json.dumps(standard_logging_payload)
+    assert standard_logging_payload["messages"][0]["content"] == "redacted-by-litellm"
 
 
 @pytest.mark.asyncio
diff --git a/tests/proxy_unit_tests/test_audit_logs_proxy.py b/tests/proxy_unit_tests/test_audit_logs_proxy.py
index 7637229336..8d420cea36 100644
--- a/tests/proxy_unit_tests/test_audit_logs_proxy.py
+++ b/tests/proxy_unit_tests/test_audit_logs_proxy.py
@@ -45,8 +45,11 @@ verbose_proxy_logger.setLevel(level=logging.DEBUG)
 
 from starlette.datastructures import URL
 
-from litellm.proxy.management_helpers.audit_logs import create_audit_log_for_update
-from litellm.proxy._types import LiteLLM_AuditLogs, LitellmTableNames
+from litellm.proxy.management_helpers.audit_logs import (
+    create_audit_log_for_update,
+    get_audit_log_changed_by,
+)
+from litellm.proxy._types import LiteLLM_AuditLogs, LitellmTableNames, UserAPIKeyAuth
 from litellm.caching.caching import DualCache
 from unittest.mock import patch, AsyncMock
 
@@ -54,6 +57,35 @@ proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
 import json
 
 
+def test_get_audit_log_changed_by_prefers_authenticated_user():
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="test-key",
+        user_id="authenticated-user",
+    )
+
+    assert (
+        get_audit_log_changed_by(
+            litellm_changed_by="spoofed-user",
+            user_api_key_dict=user_api_key_dict,
+            litellm_proxy_admin_name="proxy-admin",
+        )
+        == "authenticated-user"
+    )
+
+
+def test_get_audit_log_changed_by_falls_back_to_header_when_user_id_missing():
+    user_api_key_dict = UserAPIKeyAuth(api_key="test-key")
+
+    assert (
+        get_audit_log_changed_by(
+            litellm_changed_by="delegated-user",
+            user_api_key_dict=user_api_key_dict,
+            litellm_proxy_admin_name="proxy-admin",
+        )
+        == "delegated-user"
+    )
+
+
 @pytest.mark.asyncio
 async def test_create_audit_log_for_update_premium_user():
     """
diff --git a/tests/proxy_unit_tests/test_proxy_server.py b/tests/proxy_unit_tests/test_proxy_server.py
index c62c3f41b3..86bbc5170e 100644
--- a/tests/proxy_unit_tests/test_proxy_server.py
+++ b/tests/proxy_unit_tests/test_proxy_server.py
@@ -1553,6 +1553,7 @@ async def test_add_callback_via_key(prisma_client):
                 fastapi_response=Response(),
                 user_api_key_dict=UserAPIKeyAuth(
                     metadata={
+                        "allow_client_mock_response": True,
                         "logging": [
                             {
                                 "callback_name": "langfuse",  # 'otel', 'langfuse', 'lunary'
@@ -1563,7 +1564,7 @@ async def test_add_callback_via_key(prisma_client):
                                     "langfuse_host": "https://us.cloud.langfuse.com",
                                 },
                             }
-                        ]
+                        ],
                     }
                 ),
             )
@@ -1657,6 +1658,7 @@ async def test_add_callback_via_key_litellm_pre_call_utils(
             team_id=None,
             max_parallel_requests=None,
             metadata={
+                "allow_client_mock_response": True,
                 "logging": [
                     {
                         "callback_name": "langfuse",
@@ -1667,7 +1669,7 @@ async def test_add_callback_via_key_litellm_pre_call_utils(
                             "langfuse_host": "https://us.cloud.langfuse.com",
                         },
                     }
-                ]
+                ],
             },
             tpm_limit=None,
             rpm_limit=None,
@@ -1813,6 +1815,7 @@ async def test_add_callback_via_key_litellm_pre_call_utils_gcs_bucket(
             team_id=None,
             max_parallel_requests=None,
             metadata={
+                "allow_client_mock_response": True,
                 "logging": [
                     {
                         "callback_name": "gcs_bucket",
@@ -1822,7 +1825,7 @@ async def test_add_callback_via_key_litellm_pre_call_utils_gcs_bucket(
                             "gcs_path_service_account": "pathrise-convert-1606954137718-a956eef1a2a8.json",
                         },
                     }
-                ]
+                ],
             },
             tpm_limit=None,
             rpm_limit=None,
@@ -1946,6 +1949,7 @@ async def test_add_callback_via_key_litellm_pre_call_utils_langsmith(
             team_id=None,
             max_parallel_requests=None,
             metadata={
+                "allow_client_mock_response": True,
                 "logging": [
                     {
                         "callback_name": "langsmith",
@@ -1956,7 +1960,7 @@ async def test_add_callback_via_key_litellm_pre_call_utils_langsmith(
                             "langsmith_base_url": "https://api.smith.langchain.com",
                         },
                     }
-                ]
+                ],
             },
             tpm_limit=None,
             rpm_limit=None,
diff --git a/tests/test_litellm/litellm_core_utils/test_redact_messages.py b/tests/test_litellm/litellm_core_utils/test_redact_messages.py
index 109e344d72..58464b2fbd 100644
--- a/tests/test_litellm/litellm_core_utils/test_redact_messages.py
+++ b/tests/test_litellm/litellm_core_utils/test_redact_messages.py
@@ -68,13 +68,20 @@ class TestShouldRedactMessageLogging:
         assert should_redact_message_logging(details) is True
 
     def test_disable_redaction_via_header_proxy_flow(self):
-        """litellm-disable-message-redaction should suppress redaction
-        even when global setting is on, and litellm_metadata is None."""
+        """Global redaction should override litellm-disable-message-redaction."""
         litellm.turn_off_message_logging = True
         details = _make_model_call_details(
             metadata_headers={"litellm-disable-message-redaction": "true"},
             litellm_metadata=None,
         )
+        assert should_redact_message_logging(details) is True
+
+    def test_disable_redaction_via_header_when_global_off(self):
+        """litellm-disable-message-redaction is still honored when global redaction is off."""
+        details = _make_model_call_details(
+            metadata_headers={"litellm-disable-message-redaction": "true"},
+            litellm_metadata=None,
+        )
         assert should_redact_message_logging(details) is False
 
     # ---- SDK direct-call flow: headers in litellm_metadata ----
@@ -127,6 +134,16 @@ class TestShouldRedactMessageLogging:
         )
         assert should_redact_message_logging(details) is False
 
+    def test_global_redaction_overrides_dynamic_param_false(self):
+        """Global redaction cannot be disabled by a dynamic parameter."""
+        litellm.turn_off_message_logging = True
+        details = _make_model_call_details(
+            metadata_headers={},
+            litellm_metadata=None,
+            standard_callback_dynamic_params={"turn_off_message_logging": False},
+        )
+        assert should_redact_message_logging(details) is True
+
     # ---- non-dict metadata safety ----
 
     def test_both_metadata_fields_none(self):
diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py
index 7dfdca118c..a3247d2e55 100644
--- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py
+++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py
@@ -1853,6 +1853,60 @@ async def test_make_bedrock_api_request_logging_event_type_for_spend_logs():
         assert mock_log.call_args.kwargs["event_type"] == GuardrailEventHooks.pre_call
 
 
+@pytest.mark.asyncio
+async def test_make_bedrock_api_request_filters_dynamic_evaluation_overrides():
+    guardrail = BedrockGuardrail(
+        guardrailIdentifier="test-guardrail", guardrailVersion="DRAFT"
+    )
+    mock_credentials = MagicMock()
+    mock_credentials.access_key = "test-access-key"
+    mock_credentials.secret_key = "test-secret-key"
+    mock_credentials.token = None
+
+    mock_bedrock_response = MagicMock()
+    mock_bedrock_response.status_code = 200
+    mock_bedrock_response.json.return_value = {"action": "NONE", "assessments": []}
+
+    prepared_request = MagicMock()
+    prepared_request.url = "https://bedrock.test/apply"
+    prepared_request.body = b"{}"
+    prepared_request.headers = {}
+
+    with (
+        patch.object(
+            guardrail.async_handler, "post", new_callable=AsyncMock
+        ) as mock_post,
+        patch.object(
+            guardrail, "_load_credentials", return_value=(mock_credentials, "us-east-1")
+        ),
+        patch.object(
+            guardrail, "_prepare_request", return_value=prepared_request
+        ) as mock_prepare_request,
+        patch.object(
+            guardrail,
+            "get_guardrail_dynamic_request_body_params",
+            return_value={
+                "content": [{"text": {"text": "benign replacement"}}],
+                "source": "OUTPUT",
+                "outputScope": "FULL",
+            },
+        ),
+    ):
+        mock_post.return_value = mock_bedrock_response
+
+        await guardrail.make_bedrock_api_request(
+            source="INPUT",
+            messages=[{"role": "user", "content": "actual prompt"}],
+            request_data={"model": "gpt-4o"},
+        )
+
+    prepared_data = mock_prepare_request.call_args.kwargs["data"]
+    assert prepared_data["source"] == "INPUT"
+    assert "actual prompt" in json.dumps(prepared_data["content"])
+    assert "benign replacement" not in json.dumps(prepared_data["content"])
+    assert prepared_data["outputScope"] == "FULL"
+
+
 @pytest.mark.asyncio
 async def test_during_call_hook_invokes_bedrock_async_moderation_hook():
     """
diff --git a/tests/test_litellm/proxy/guardrails/test_pillar_guardrails.py b/tests/test_litellm/proxy/guardrails/test_pillar_guardrails.py
index 6f9e030269..c977223bfa 100644
--- a/tests/test_litellm/proxy/guardrails/test_pillar_guardrails.py
+++ b/tests/test_litellm/proxy/guardrails/test_pillar_guardrails.py
@@ -505,6 +505,38 @@ def test_get_logging_caching_headers_pillar_metadata():
     )
 
 
+def test_get_logging_caching_headers_ignores_untrusted_pillar_headers():
+    request_data = {
+        "metadata": {
+            "pillar_response_headers": {
+                "set-cookie": "session=evil",
+                "x-pillar-flagged": "true",
+            },
+            "pillar_flagged": True,
+        }
+    }
+
+    headers = get_logging_caching_headers(request_data)
+
+    assert "set-cookie" not in headers
+    assert "x-pillar-flagged" not in headers
+
+
+def test_get_logging_caching_headers_filters_non_pillar_headers():
+    request_data = {
+        "metadata": {
+            "pillar_flagged": True,
+        }
+    }
+    build_pillar_response_headers(request_data["metadata"])
+    request_data["metadata"]["pillar_response_headers"]["set-cookie"] = "session=evil"
+
+    headers = get_logging_caching_headers(request_data)
+
+    assert headers["x-pillar-flagged"] == "true"
+    assert "set-cookie" not in headers
+
+
 def test_get_logging_caching_headers_truncates_large_evidence():
     long_text = "悪" * 6000  # multi-byte unicode to test URL encoding and truncation
     request_data = {
diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
index 1e5c688e36..d06740745c 100644
--- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
+++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
@@ -512,6 +512,122 @@ async def test_add_litellm_data_to_request_strips_string_encoded_admin_injection
     assert "_pipeline_managed_guardrails" not in other
 
 
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_strips_user_control_fields():
+    """Strip untrusted proxy-control fields before guardrails, logging, and headers read metadata."""
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    malicious_metadata = {
+        "disable_global_guardrails": True,
+        "opted_out_global_guardrails": ["pii"],
+        "pillar_response_headers": {"set-cookie": "session=evil"},
+        "_pillar_response_headers_trusted": True,
+        "pillar_flagged": True,
+        "pillar_scanners": {"jailbreak": True},
+        "pillar_evidence": [{"evidence": "spoofed"}],
+        "pillar_session_id_response": "spoofed-session",
+        "applied_guardrails": ["spoofed"],
+        "applied_policies": ["spoofed-policy"],
+        "policy_sources": {"spoofed-policy": "request"},
+        "_guardrail_pipelines": [{"name": "spoofed"}],
+        "_pipeline_managed_guardrails": ["evaded"],
+        "safe_user_metadata": "kept",
+    }
+    data = {
+        "model": "gpt-3.5-turbo",
+        "messages": [{"role": "user", "content": "hello"}],
+        "mock_response": "free response",
+        "mock_tool_calls": [{"id": "call_1"}],
+        "disable_global_guardrails": True,
+        "metadata": copy.deepcopy(malicious_metadata),
+        "litellm_metadata": copy.deepcopy(malicious_metadata),
+    }
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key"),
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert "mock_response" not in updated
+    assert "mock_tool_calls" not in updated
+    assert "disable_global_guardrails" not in updated
+
+    stripped_keys = {
+        "disable_global_guardrails",
+        "opted_out_global_guardrails",
+        "pillar_response_headers",
+        "_pillar_response_headers_trusted",
+        "pillar_flagged",
+        "pillar_scanners",
+        "pillar_evidence",
+        "pillar_session_id_response",
+        "applied_guardrails",
+        "applied_policies",
+        "policy_sources",
+        "_guardrail_pipelines",
+        "_pipeline_managed_guardrails",
+    }
+    for metadata_key in ("metadata", "litellm_metadata"):
+        cleaned_metadata = updated.get(metadata_key) or {}
+        for stripped_key in stripped_keys:
+            assert stripped_key not in cleaned_metadata
+        assert cleaned_metadata.get("safe_user_metadata") == "kept"
+
+    requester_metadata = updated["metadata"]["requester_metadata"]
+    for stripped_key in stripped_keys:
+        assert stripped_key not in requester_metadata
+
+    snapshot_body = updated["proxy_server_request"]["body"]
+    assert "mock_response" not in snapshot_body
+    assert "mock_tool_calls" not in snapshot_body
+    assert "pillar_response_headers" not in snapshot_body["metadata"]
+
+
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_allows_client_mock_response_with_admin_opt_in():
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    updated = await add_litellm_data_to_request(
+        data={
+            "model": "gpt-3.5-turbo",
+            "messages": [{"role": "user", "content": "hello"}],
+            "mock_response": "allowed mock",
+            "mock_tool_calls": [{"id": "call_1"}],
+        },
+        request=request_mock,
+        user_api_key_dict=UserAPIKeyAuth(
+            api_key="hashed-key",
+            metadata={"allow_client_mock_response": True},
+        ),
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert updated["mock_response"] == "allowed mock"
+    assert updated["mock_tool_calls"] == [{"id": "call_1"}]
+
+
 @pytest.mark.asyncio
 async def test_add_litellm_data_to_request_ignores_x_litellm_tags_header_without_permission():
     """Regression: the `x-litellm-tags` header bypassed the body-metadata
@@ -3326,7 +3442,9 @@ async def test_team_guardrail_merges_with_global_policy():
     policy_registry = get_policy_registry()
     policy_registry._policies = {
         "global-policy": Policy(
-            guardrails=PolicyGuardrails(add=["policy-guardrail-1", "policy-guardrail-2"]),
+            guardrails=PolicyGuardrails(
+                add=["policy-guardrail-1", "policy-guardrail-2"]
+            ),
         ),
     }
     policy_registry._initialized = True
@@ -3347,14 +3465,18 @@ async def test_team_guardrail_merges_with_global_policy():
 
         guardrails = data["metadata"].get("guardrails", [])
 
-        assert "team-direct-guardrail" in guardrails, \
-            f"Team guardrail missing from merged list: {guardrails}"
-        assert "policy-guardrail-1" in guardrails, \
-            f"policy-guardrail-1 missing: {guardrails}"
-        assert "policy-guardrail-2" in guardrails, \
-            f"policy-guardrail-2 missing: {guardrails}"
-        assert len(guardrails) == len(set(guardrails)), \
-            f"Duplicates in guardrails list: {guardrails}"
+        assert (
+            "team-direct-guardrail" in guardrails
+        ), f"Team guardrail missing from merged list: {guardrails}"
+        assert (
+            "policy-guardrail-1" in guardrails
+        ), f"policy-guardrail-1 missing: {guardrails}"
+        assert (
+            "policy-guardrail-2" in guardrails
+        ), f"policy-guardrail-2 missing: {guardrails}"
+        assert len(guardrails) == len(
+            set(guardrails)
+        ), f"Duplicates in guardrails list: {guardrails}"
 
         # Verify get_guardrail_from_metadata returns the merged list even
         # when litellm_metadata is present (the bug: it returned [] before fix)
@@ -3365,9 +3487,9 @@ async def test_team_guardrail_merges_with_global_policy():
 
         dummy = _DummyGuardrail(guardrail_name="team-direct-guardrail")
         returned = dummy.get_guardrail_from_metadata(data)
-        assert "team-direct-guardrail" in returned, (
-            f"get_guardrail_from_metadata shadowed by litellm_metadata; got: {returned}"
-        )
+        assert (
+            "team-direct-guardrail" in returned
+        ), f"get_guardrail_from_metadata shadowed by litellm_metadata; got: {returned}"
 
     finally:
         policy_registry._policies = {}
@@ -3396,9 +3518,10 @@ async def test_get_guardrail_from_metadata_prefers_metadata_over_litellm_metadat
     }
 
     result = dummy.get_guardrail_from_metadata(data)
-    assert result == ["my-guardrail", "other-guardrail"], (
-        f"Expected guardrails from metadata, got: {result}"
-    )
+    assert result == [
+        "my-guardrail",
+        "other-guardrail",
+    ], f"Expected guardrails from metadata, got: {result}"
 
 
 def test_get_guardrail_from_metadata_reads_litellm_metadata_when_no_metadata():
@@ -3419,6 +3542,6 @@ def test_get_guardrail_from_metadata_reads_litellm_metadata_when_no_metadata():
     }
 
     result = dummy.get_guardrail_from_metadata(data)
-    assert result == ["my-guardrail"], (
-        f"Expected guardrails from litellm_metadata fallback, got: {result}"
-    )
+    assert result == [
+        "my-guardrail"
+    ], f"Expected guardrails from litellm_metadata fallback, got: {result}"