diff --git a/README.md b/README.md
index 531a50e1aca..1061ab793e5 100644
--- a/README.md
+++ b/README.md
@@ -141,6 +141,24 @@ response = client.chat.completions.create(
 )
 ```
 
+#### Debug logs without message content
+
+To troubleshoot with detailed logging, add these settings to your existing proxy configuration and restart with `litellm --config config.yaml --detailed_debug`
+
+```yaml
+litellm_settings:
+  redact_messages_in_debug_logs: true
+  turn_off_message_logging: true
+```
+
+`redact_messages_in_debug_logs` defaults to `false`. When enabled, LiteLLM replaces free-form diagnostic messages, payloads, and exception text with `redacted-by-litellm`, retaining selected operational events. This also applies to payload diagnostics emitted at warning or error level. Raw tracebacks and caller-supplied correlation identifiers are omitted because they can contain message content
+
+The setting affects diagnostics only. Requests sent to providers and responses returned to clients stay intact. `turn_off_message_logging` separately controls message content sent to logging integrations. Request-level callback overrides do not disable diagnostic redaction. Custom application loggers and independently configured third-party wire logging need their own privacy controls
+
+The built-in plain and JSON formatters support this mode. Custom formatters must use standard log fields or supply defaults for extra fields, which are removed to prevent content disclosure. This Python SDK/proxy setting does not configure the standalone Rust gateway
+
+Verify with a synthetic prompt: the client should receive its normal response while LiteLLM diagnostics show `redacted-by-litellm` instead of the prompt or reply. Remove the new setting or set it to `false`, then restart the proxy to restore existing diagnostics
+
 </details>
 
 <details>
diff --git a/litellm/__init__.py b/litellm/__init__.py
index 084a4c516ce..576dcc891be 100644
--- a/litellm/__init__.py
+++ b/litellm/__init__.py
@@ -189,6 +189,7 @@ _async_failure_callback: List[Union[str, Callable[..., object], "CustomLogger"]]
 pre_call_rules: List[Callable] = []
 post_call_rules: List[Callable[..., object]] = []
 turn_off_message_logging: Optional[bool] = False
+redact_messages_in_debug_logs: bool = False
 standard_logging_payload_excluded_fields: Optional[List[str]] = (
     None  # Fields to exclude from StandardLoggingPayload before callbacks receive it
 )
diff --git a/litellm/_logging.py b/litellm/_logging.py
index 6a38eea2b12..a1d2b1e2b08 100644
--- a/litellm/_logging.py
+++ b/litellm/_logging.py
@@ -7,11 +7,12 @@ import logging
 import os
 import re
 import sys
+import traceback
 from collections.abc import Iterator, Mapping
 from datetime import datetime
 from logging import Formatter
 from types import FrameType
-from typing import Final, TextIO
+from typing import Final, Literal, TextIO
 from urllib.parse import unquote
 
 from pydantic import TypeAdapter
@@ -23,6 +24,7 @@ from litellm.constants import (
     LITELLM_TRUNCATION_STDOUT_SAFEGUARD_NOTE,
     MAX_BASE64_LENGTH_STDOUT_LOG,
     MAX_STRING_LENGTH_STDOUT_LOG,
+    REDACTED_BY_LITELLM,
 )
 from litellm.litellm_core_utils.env_utils import get_env_int
 from litellm.litellm_core_utils.safe_json_dumps import UNSERIALIZABLE_OBJECT, safe_dumps, safe_json_structure
@@ -91,6 +93,88 @@ _redact_string = redact_string
 
 _REDACTED_RECORD_ATTR: Final = "litellm_redacted"
 _REDACTED_STAMP: Final = object()
+_SAFE_DIAGNOSTIC_ATTR: Final = "litellm_safe_diagnostic"
+_SAFE_DIAGNOSTIC_STAMP: Final = object()
+
+
+def diagnostic_message_logging_disabled() -> bool:
+    return getattr(litellm, "redact_messages_in_debug_logs", False) is True
+
+
+def _litellm_exception_record(record: logging.LogRecord) -> bool:
+    if not isinstance(record.exc_info, tuple) or record.exc_info[2] is None:
+        return False
+    return any(
+        isinstance(module := frame.f_globals.get("__name__"), str)
+        and (module == "litellm" or module.startswith("litellm."))
+        for frame, _ in traceback.walk_tb(record.exc_info[2])
+    )
+
+
+def _redact_diagnostic_record(record: logging.LogRecord) -> None:
+    if (
+        not diagnostic_message_logging_disabled()
+        or getattr(record, _SAFE_DIAGNOSTIC_ATTR, None) is _SAFE_DIAGNOSTIC_STAMP
+    ):
+        return
+    owned: Final = record.name in (
+        "LiteLLM",
+        "LiteLLM Router",
+        "LiteLLM Proxy",
+        "uvicorn.error",
+    ) or record.name.startswith(("LiteLLM.", "LiteLLM Router.", "LiteLLM Proxy.", "litellm."))
+    if not owned and not (record.name == "asyncio" and _litellm_exception_record(record)):
+        return
+    record.msg = REDACTED_BY_LITELLM  # rebind-ok: logging filters sanitize the diagnostic copy
+    record.args = ()  # rebind-ok: never render payload arguments
+    record.exc_info = None  # rebind-ok: exception messages may echo request content
+    record.exc_text = None  # rebind-ok: remove previously rendered exception content
+    record.stack_info = None  # rebind-ok: stack text may contain payloads
+    for key in tuple(record.__dict__):
+        if key not in _STANDARD_RECORD_ATTRS:
+            delattr(record, key)
+    setattr(record, _REDACTED_RECORD_ATTR, _REDACTED_STAMP)
+
+
+class MessageContentFilter(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        _redact_diagnostic_record(record)
+        return True
+
+
+def log_safe_diagnostic(
+    logger: logging.Logger,
+    event: Literal[
+        "request_sent",
+        "response_received",
+        "response_ready",
+        "request_completed",
+        "request_failed",
+        "stream_completed",
+        "retry",
+        "fallback",
+    ],
+    *,
+    status_code: int | None = None,
+    elapsed_seconds: float | None = None,
+    attempt: int | None = None,
+    level: int = logging.DEBUG,
+) -> None:
+    if not diagnostic_message_logging_disabled() or not logger.isEnabledFor(level):
+        return
+    logger.log(
+        level,
+        "event=%s status=%s elapsed_seconds=%s attempt=%s content=%s",
+        event,
+        status_code,
+        elapsed_seconds,
+        attempt,
+        REDACTED_BY_LITELLM,
+        extra={_SAFE_DIAGNOSTIC_ATTR: _SAFE_DIAGNOSTIC_STAMP},
+        stacklevel=2,
+    )
+
+
 _DECODED_LOG_PAYLOAD: Final = TypeAdapter(object)
 _UNREDACTED_SCALAR_TYPES: Final = (bool, int, float, type(None))
 
@@ -192,6 +276,7 @@ class SecretRedactionFilter(logging.Filter):
     _formatter = logging.Formatter()
 
     def filter(self, record: logging.LogRecord) -> bool:
+        _redact_diagnostic_record(record)
         if not _ENABLE_SECRET_REDACTION or _is_redacted(record):
             return True
         return _process_record(record, base64_limit=0, text_limit=0, redact=True)
@@ -252,6 +337,23 @@ class AccessLogRedactionFilter(logging.Filter):
     """
 
     def filter(self, record: logging.LogRecord) -> bool:
+        if diagnostic_message_logging_disabled():
+            safe_args: Final = (
+                tuple(value if isinstance(value, (int, float, bool)) else REDACTED_BY_LITELLM for value in record.args)
+                if isinstance(record.args, tuple) and len(record.args) == 5
+                else ()
+            )
+            record.msg = (  # rebind-ok: access diagnostic copy
+                '%s - "%s %s HTTP/%s" %s' if safe_args else REDACTED_BY_LITELLM
+            )
+            record.args = safe_args  # rebind-ok: preserve the access formatter's tuple contract
+            record.exc_info = None  # rebind-ok: exceptions can echo request content
+            record.exc_text = None  # rebind-ok: discard cached exception content
+            record.stack_info = None  # rebind-ok: discard stack content
+            for key in tuple(record.__dict__):
+                if key not in _STANDARD_RECORD_ATTRS:
+                    delattr(record, key)
+            return True
         if not _ENABLE_SECRET_REDACTION:
             return True
         if isinstance(record.args, tuple) and record.args:
@@ -461,6 +563,7 @@ def _render_exception(record: logging.LogRecord) -> str | None:
 
 
 def _process_record(record: logging.LogRecord, *, base64_limit: int, text_limit: int, redact: bool) -> bool:
+    _redact_diagnostic_record(record)
     if _is_redacted(record):
         return True
     message: Final = _render_message(record)
@@ -478,7 +581,7 @@ def _process_record(record: logging.LogRecord, *, base64_limit: int, text_limit:
             )
             for key, value in _record_attributes(record).items()
             if key not in _STANDARD_RECORD_ATTRS
-            and key != _REDACTED_RECORD_ATTR
+            and key not in (_REDACTED_RECORD_ATTR, _SAFE_DIAGNOSTIC_ATTR)
             and not isinstance(value, _UNREDACTED_SCALAR_TYPES)
         )
         if redact
@@ -562,6 +665,7 @@ class StdoutLogTruncationFilter(logging.Filter):
     _formatter = logging.Formatter()
 
     def filter(self, record: logging.LogRecord) -> bool:
+        _redact_diagnostic_record(record)
         try:
             message: Final = record.getMessage()
         except (TypeError, ValueError):
@@ -610,7 +714,7 @@ class CorrelationContextFilter(logging.Filter):
     """
 
     def filter(self, record: logging.LogRecord) -> bool:
-        if not litellm.request_correlation_in_logs:
+        if diagnostic_message_logging_disabled() or not litellm.request_correlation_in_logs:
             return True
         trace_id: Final = trace_id_var.get()
         if trace_id:
@@ -774,13 +878,13 @@ def _try_parse_embedded_python_dict(message: str) -> dict[str, object] | None:
 
 # Standard LogRecord attribute names - used to identify 'extra' fields.
 # Derived at runtime so we automatically include version-specific attrs (e.g. taskName).
-def _get_standard_record_attrs() -> frozenset:
+def _get_standard_record_attrs() -> frozenset[str]:
     """Standard LogRecord attribute names - excludes extra keys from logger.debug(..., extra={...})."""
     return frozenset(logging.LogRecord("", 0, "", 0, "", (), None).__dict__.keys())
 
 
 _STANDARD_RECORD_ATTRS: Final = _get_standard_record_attrs()
-_NON_EXTRA_RECORD_ATTRS: Final = _STANDARD_RECORD_ATTRS | {_REDACTED_RECORD_ATTR}
+_NON_EXTRA_RECORD_ATTRS: Final = _STANDARD_RECORD_ATTRS | {_REDACTED_RECORD_ATTR, _SAFE_DIAGNOSTIC_ATTR}
 
 # CorrelationContextFilter is the only legitimate source for these two JSON fields;
 # see JsonFormatter.format() for why they're excluded from the generic message-content
@@ -798,6 +902,7 @@ class JsonFormatter(Formatter):
         return dt.isoformat()
 
     def format(self, record):
+        _redact_diagnostic_record(record)
         message_str: Final = _ANSI_ESCAPE.sub("", record.getMessage())
         json_record: Final[dict[str, object]] = {
             "message": message_str,
@@ -890,7 +995,7 @@ def _setup_json_exception_handlers(formatter):
             level=logging.ERROR,
             pathname="",
             lineno=0,
-            msg=str(exc_value),
+            msg=REDACTED_BY_LITELLM if diagnostic_message_logging_disabled() else str(exc_value),
             args=(),
             exc_info=(exc_type, exc_value, exc_traceback),
         )
@@ -911,7 +1016,7 @@ def _setup_json_exception_handlers(formatter):
                     level=logging.ERROR,
                     pathname="",
                     lineno=0,
-                    msg=str(exception),
+                    msg=REDACTED_BY_LITELLM if diagnostic_message_logging_disabled() else str(exception),
                     args=(),
                     exc_info=(exc_type, exception, exception.__traceback__),
                 )
@@ -987,7 +1092,7 @@ def _write_startup_message(message: str, caller: FrameType | None, level: int) -
     if not litellm.json_logs:
         import click
 
-        click.echo(terminal_text(message))
+        click.echo(REDACTED_BY_LITELLM if diagnostic_message_logging_disabled() else terminal_text(message))
         return
     text: Final = _ANSI_ESCAPE.sub("", message).strip("\n")
     if not text:
@@ -1227,7 +1332,13 @@ def print_verbose(print_statement: object, *args: object) -> None:
     try:
         if not set_verbose:
             return
-        message: Final = str(print_statement) % args if args else str(print_statement)
+        message: Final = (
+            REDACTED_BY_LITELLM
+            if diagnostic_message_logging_disabled()
+            else str(print_statement) % args
+            if args
+            else str(print_statement)
+        )
         print(redact_secrets(message))  # noqa: T201
     except Exception:
         pass
diff --git a/litellm/budget_manager.py b/litellm/budget_manager.py
index 0f8ec4be036..e133f7e3863 100644
--- a/litellm/budget_manager.py
+++ b/litellm/budget_manager.py
@@ -47,7 +47,7 @@ class BudgetManager:
             if litellm.set_verbose:
                 import logging
 
-                logging.info(print_statement)
+                logging.info("redacted-by-litellm" if litellm.redact_messages_in_debug_logs else print_statement)
         except Exception:
             pass
 
diff --git a/litellm/caching/caching.py b/litellm/caching/caching.py
index d875ac09d40..481431ef78d 100644
--- a/litellm/caching/caching.py
+++ b/litellm/caching/caching.py
@@ -61,7 +61,7 @@ def print_verbose(print_statement):
     try:
         verbose_logger.debug(print_statement)
         if litellm.set_verbose:
-            print(print_statement)  # noqa: T201
+            print("redacted-by-litellm" if litellm.redact_messages_in_debug_logs else print_statement)  # noqa: T201
     except Exception:
         pass
 
diff --git a/litellm/caching/dual_cache.py b/litellm/caching/dual_cache.py
index e6214a3650c..5be1bd3da1e 100644
--- a/litellm/caching/dual_cache.py
+++ b/litellm/caching/dual_cache.py
@@ -198,7 +198,9 @@ class DualCache(BaseCache):
 
                 result = redis_result
 
-            print_verbose(f"get cache: cache result: {result}")
+            print_verbose(
+                "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"get cache: cache result: {result}"
+            )
             return result
         except Exception as e:
             log_redis_failure(
@@ -262,7 +264,11 @@ class DualCache(BaseCache):
             if self.in_memory_cache is not None:
                 in_memory_result: Final = await self.in_memory_cache.async_get_cache(key, **kwargs)
 
-                print_verbose(f"in_memory_result: {in_memory_result}")
+                print_verbose(
+                    "redacted-by-litellm"
+                    if litellm.redact_messages_in_debug_logs
+                    else f"in_memory_result: {in_memory_result}"
+                )
                 if in_memory_result is not None:
                     result = in_memory_result
 
@@ -279,7 +285,9 @@ class DualCache(BaseCache):
 
                 result = redis_result
 
-            print_verbose(f"get cache: cache result: {result}")
+            print_verbose(
+                "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"get cache: cache result: {result}"
+            )
             return result
         except Exception as e:
             log_redis_failure(
diff --git a/litellm/caching/qdrant_semantic_cache.py b/litellm/caching/qdrant_semantic_cache.py
index 560a5ba06b8..22a5d01a0e7 100644
--- a/litellm/caching/qdrant_semantic_cache.py
+++ b/litellm/caching/qdrant_semantic_cache.py
@@ -208,7 +208,11 @@ class QdrantSemanticCache(BaseCache):
                 },
             )
             if response.status_code not in (200, 201):
-                print_verbose(f"Qdrant semantic-cache could not create cache-key payload index: {response.text}")
+                print_verbose(
+                    "redacted-by-litellm"
+                    if litellm.redact_messages_in_debug_logs
+                    else f"Qdrant semantic-cache could not create cache-key payload index: {response.text}"
+                )
         except Exception as exc:
             print_verbose(f"Qdrant semantic-cache could not create cache-key payload index: {exc}")
 
@@ -284,7 +288,11 @@ class QdrantSemanticCache(BaseCache):
         return await asyncio.wait_for(embedding_call, self.embedding_timeout)
 
     def set_cache(self, key, value, **kwargs):
-        print_verbose(f"qdrant semantic-cache set_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"qdrant semantic-cache set_cache, kwargs: {kwargs}"
+        )
         from litellm._uuid import uuid
 
         # get the prompt
@@ -321,7 +329,11 @@ class QdrantSemanticCache(BaseCache):
         )
 
     def get_cache(self, key, **kwargs):
-        print_verbose(f"sync qdrant semantic-cache get_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"sync qdrant semantic-cache get_cache, kwargs: {kwargs}"
+        )
 
         # get the messages
         messages: Final = kwargs["messages"]
@@ -393,7 +405,11 @@ class QdrantSemanticCache(BaseCache):
     async def async_set_cache(self, key, value, **kwargs):
         from litellm._uuid import uuid
 
-        print_verbose(f"async qdrant semantic-cache set_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"async qdrant semantic-cache set_cache, kwargs: {kwargs}"
+        )
 
         # get the prompt
         messages: Final = kwargs["messages"]
@@ -428,7 +444,11 @@ class QdrantSemanticCache(BaseCache):
         )
 
     async def async_get_cache(self, key, **kwargs):
-        print_verbose(f"async qdrant semantic-cache get_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"async qdrant semantic-cache get_cache, kwargs: {kwargs}"
+        )
 
         # get the messages
         messages: Final = kwargs["messages"]
diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py
index afe6f8a1eab..bc379953542 100644
--- a/litellm/caching/redis_cache.py
+++ b/litellm/caching/redis_cache.py
@@ -1266,7 +1266,9 @@ class RedisCache(BaseCache):
             async with _redis_client.pipeline(transaction=False) as pipe:
                 results: Final = await self._pipeline_helper(pipe, cache_list, ttl)
 
-            print_verbose(f"pipeline results: {results}")
+            print_verbose(
+                "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"pipeline results: {results}"
+            )
             # Optionally, you could process 'results' to make sure that all set operations were successful.
             ## LOGGING ##
             end_time = time.time()
@@ -1870,7 +1872,9 @@ class RedisCache(BaseCache):
         start_time: Final = time.time()
         try:
             response: Final[bool] = self.redis_client.ping()
-            print_verbose(f"Redis Cache PING: {response}")
+            print_verbose(
+                "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"Redis Cache PING: {response}"
+            )
             ## LOGGING ##
             end_time = time.time()
             _duration = end_time - start_time
diff --git a/litellm/caching/redis_semantic_cache.py b/litellm/caching/redis_semantic_cache.py
index 7268511fe7e..a0ec353371e 100644
--- a/litellm/caching/redis_semantic_cache.py
+++ b/litellm/caching/redis_semantic_cache.py
@@ -425,7 +425,11 @@ class RedisSemanticCache(BaseCache):
             **kwargs: Additional arguments including 'messages' for the prompt
                 and optional 'ttl' for time-to-live
         """
-        print_verbose(f"Redis semantic-cache set_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Redis semantic-cache set_cache, kwargs: {kwargs}"
+        )
 
         value_str: str | None = None
         try:
@@ -462,7 +466,11 @@ class RedisSemanticCache(BaseCache):
         Returns:
             The cached response if a semantically similar prompt is found, else None
         """
-        print_verbose(f"Redis semantic-cache get_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Redis semantic-cache get_cache, kwargs: {kwargs}"
+        )
 
         try:
             prompt: Final = self._get_prompt_from_kwargs(**kwargs)
@@ -571,7 +579,11 @@ class RedisSemanticCache(BaseCache):
             **kwargs: Additional arguments including 'messages' for the prompt
                 and optional 'ttl' for time-to-live
         """
-        print_verbose(f"Async Redis semantic-cache set_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Async Redis semantic-cache set_cache, kwargs: {kwargs}"
+        )
 
         try:
             prompt: Final = self._get_prompt_from_kwargs(**kwargs)
@@ -612,7 +624,11 @@ class RedisSemanticCache(BaseCache):
         Returns:
             The cached response if a semantically similar prompt is found, else None
         """
-        print_verbose(f"Async Redis semantic-cache get_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Async Redis semantic-cache get_cache, kwargs: {kwargs}"
+        )
 
         try:
             prompt: Final = self._get_prompt_from_kwargs(**kwargs)
diff --git a/litellm/caching/valkey_semantic_cache.py b/litellm/caching/valkey_semantic_cache.py
index 645bd0892f4..e6e67a35338 100644
--- a/litellm/caching/valkey_semantic_cache.py
+++ b/litellm/caching/valkey_semantic_cache.py
@@ -28,6 +28,7 @@ from redis.commands.search.field import TagField, VectorField
 from redis.commands.search.indexDefinition import IndexDefinition, IndexType
 from redis.commands.search.query import Query
 
+import litellm
 from litellm._logging import print_verbose
 from litellm._uuid import uuid
 from litellm.llms.valkey.common_utils import build_valkey_url, pack_vector
@@ -259,7 +260,11 @@ class ValkeySemanticCache(RedisSemanticCache):
         return self._get_cache_logic(cached_response=hit.response)
 
     def set_cache(self, key: str, value: object, **kwargs: object) -> None:
-        print_verbose(f"Valkey semantic-cache set_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Valkey semantic-cache set_cache, kwargs: {kwargs}"
+        )
         try:
             prompt: Final = self._get_prompt_from_kwargs(**kwargs)
             if prompt is None:
@@ -278,7 +283,11 @@ class ValkeySemanticCache(RedisSemanticCache):
             print_verbose(f"Error in Valkey semantic-cache set_cache: {e}")
 
     def get_cache(self, key: str, **kwargs: object) -> object:
-        print_verbose(f"Valkey semantic-cache get_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Valkey semantic-cache get_cache, kwargs: {kwargs}"
+        )
         try:
             prompt: Final = self._get_prompt_from_kwargs(**kwargs)
             if prompt is None:
@@ -298,7 +307,11 @@ class ValkeySemanticCache(RedisSemanticCache):
             self._record_similarity(kwargs, 0.0)
 
     async def async_set_cache(self, key: str, value: object, **kwargs: object) -> None:
-        print_verbose(f"Async Valkey semantic-cache set_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Async Valkey semantic-cache set_cache, kwargs: {kwargs}"
+        )
         try:
             prompt: Final = self._get_prompt_from_kwargs(**kwargs)
             if prompt is None:
@@ -317,7 +330,11 @@ class ValkeySemanticCache(RedisSemanticCache):
             print_verbose(f"Error in async Valkey semantic-cache set_cache: {e}")
 
     async def async_get_cache(self, key: str, **kwargs: object) -> object:
-        print_verbose(f"Async Valkey semantic-cache get_cache, kwargs: {kwargs}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Async Valkey semantic-cache get_cache, kwargs: {kwargs}"
+        )
         try:
             prompt: Final = self._get_prompt_from_kwargs(**kwargs)
             if prompt is None:
diff --git a/litellm/integrations/SlackAlerting/slack_alerting.py b/litellm/integrations/SlackAlerting/slack_alerting.py
index dd611947228..5c6e9427849 100644
--- a/litellm/integrations/SlackAlerting/slack_alerting.py
+++ b/litellm/integrations/SlackAlerting/slack_alerting.py
@@ -1221,7 +1221,10 @@ Model Info:
         if response.status_code == 200:
             return True
         else:
-            print("Error sending webhook alert. Error=", response.text)  # noqa: T201
+            print(  # noqa: T201  # legacy stdout diagnostic
+                "Error sending webhook alert. Error=",
+                "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else response.text,
+            )
 
         return False
 
diff --git a/litellm/integrations/cloudzero/cz_stream_api.py b/litellm/integrations/cloudzero/cz_stream_api.py
index d0b25ea83e8..00bb7381100 100644
--- a/litellm/integrations/cloudzero/cz_stream_api.py
+++ b/litellm/integrations/cloudzero/cz_stream_api.py
@@ -27,6 +27,8 @@ import httpx
 import polars as pl
 from rich.console import Console
 
+import litellm
+
 
 class CloudZeroStreamer:
     """Stream CBF data to CloudZero AnyCost API with proper batching and timezone handling."""
@@ -44,7 +46,11 @@ class CloudZeroStreamer:
             try:
                 self.user_timezone = zoneinfo.ZoneInfo(user_timezone)
             except zoneinfo.ZoneInfoNotFoundError:
-                self.console.print(f"[yellow]Warning: Unknown timezone '{user_timezone}', using UTC[/yellow]")
+                self.console.print(
+                    "redacted-by-litellm"
+                    if litellm.redact_messages_in_debug_logs
+                    else f"[yellow]Warning: Unknown timezone '{user_timezone}', using UTC[/yellow]"
+                )
                 self.user_timezone = timezone.utc
         else:
             self.user_timezone = timezone.utc
@@ -62,7 +68,11 @@ class CloudZeroStreamer:
             self.console.print("[yellow]No valid daily batches to send[/yellow]")
             return
 
-        self.console.print(f"[blue]Sending {len(daily_batches)} daily batch(es) with operation '{operation}'[/blue]")
+        self.console.print(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"[blue]Sending {len(daily_batches)} daily batch(es) with operation '{operation}'[/blue]"
+        )
 
         for batch_date, batch_data in daily_batches.items():
             self._send_daily_batch(batch_date, batch_data, operation)
@@ -94,7 +104,11 @@ class CloudZeroStreamer:
                 daily_batches[batch_date].append(row)
 
             except Exception as e:
-                self.console.print(f"[yellow]Warning: Could not process timestamp '{timestamp_str}': {e}[/yellow]")
+                self.console.print(
+                    "redacted-by-litellm"
+                    if litellm.redact_messages_in_debug_logs
+                    else f"[yellow]Warning: Could not process timestamp '{timestamp_str}': {e}[/yellow]"
+                )
                 continue
 
         # Convert lists back to DataFrames
@@ -171,21 +185,33 @@ class CloudZeroStreamer:
 
         try:
             with httpx.Client(timeout=30.0) as client:
-                self.console.print(f"[blue]Sending batch for {batch_date} ({len(batch_data)} records)[/blue]")
+                self.console.print(
+                    "redacted-by-litellm"
+                    if litellm.redact_messages_in_debug_logs
+                    else f"[blue]Sending batch for {batch_date} ({len(batch_data)} records)[/blue]"
+                )
 
                 response: Final = client.post(url, headers=headers, json=payload)
                 response.raise_for_status()
 
                 self.console.print(
-                    f"[green]✓ Successfully sent batch for {batch_date} ({len(batch_data)} records)[/green]"
+                    "redacted-by-litellm"
+                    if litellm.redact_messages_in_debug_logs
+                    else f"[green]✓ Successfully sent batch for {batch_date} ({len(batch_data)} records)[/green]"
                 )
 
         except httpx.RequestError as e:
-            self.console.print(f"[red]✗ Network error sending batch for {batch_date}: {e}[/red]")
+            self.console.print(
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"[red]✗ Network error sending batch for {batch_date}: {e}[/red]"
+            )
             raise
         except httpx.HTTPStatusError as e:
             self.console.print(
-                f"[red]✗ HTTP error sending batch for {batch_date}: {e.response.status_code} {e.response.text}[/red]"
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"[red]✗ HTTP error sending batch for {batch_date}: {e.response.status_code} {e.response.text}[/red]"
             )
             raise
 
@@ -237,7 +263,11 @@ class CloudZeroStreamer:
             return api_record
 
         except Exception as e:
-            self.console.print(f"[yellow]Warning: Could not convert record to API format: {e}[/yellow]")
+            self.console.print(
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"[yellow]Warning: Could not convert record to API format: {e}[/yellow]"
+            )
             return None
 
     def _ensure_utc_timestamp(self, timestamp_str: str) -> str:
diff --git a/litellm/integrations/deepeval/api.py b/litellm/integrations/deepeval/api.py
index 60639fe2941..cc1c12b16bb 100644
--- a/litellm/integrations/deepeval/api.py
+++ b/litellm/integrations/deepeval/api.py
@@ -5,6 +5,7 @@ from typing import Final
 
 import httpx
 
+import litellm
 from litellm._logging import verbose_logger
 
 DEEPEVAL_BASE_URL: Final = "https://deepeval.confident-ai.com"
@@ -22,9 +23,13 @@ from litellm.llms.custom_httpx.http_handler import (
 
 def log_retry_error(details):
     exception: Final = details.get("exception")
-    tries: Final = details.get("tries")
+    tries: Final = "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else details.get("tries")
     if exception:
-        logging.error("Confident AI Error: %s. Retrying: %s time(s)...", exception, tries)
+        logging.error(
+            "Confident AI Error: %s. Retrying: %s time(s)...",
+            "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else exception,
+            tries,
+        )
     else:
         logging.error("Retrying: %s time(s)...", tries)
 
diff --git a/litellm/integrations/otel/plumbing/otlp_json.py b/litellm/integrations/otel/plumbing/otlp_json.py
index fc0351d5049..0588ed1a292 100644
--- a/litellm/integrations/otel/plumbing/otlp_json.py
+++ b/litellm/integrations/otel/plumbing/otlp_json.py
@@ -20,7 +20,10 @@ from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExportResult
 from opentelemetry.semconv.attributes.http_attributes import HTTP_RESPONSE_STATUS_CODE
 
+from litellm._logging import MessageContentFilter
+
 _logger: Final = logging.getLogger(__name__)
+_logger.addFilter(MessageContentFilter())
 JSON_CONTENT_TYPE: Final = "application/json"
 _HEX_ID_KEYS: Final = frozenset({"traceId", "spanId", "parentSpanId"})
 
diff --git a/litellm/integrations/weights_biases.py b/litellm/integrations/weights_biases.py
index 9bc070a1f9a..623c468a300 100644
--- a/litellm/integrations/weights_biases.py
+++ b/litellm/integrations/weights_biases.py
@@ -9,7 +9,10 @@ try:
 
     from wandb.sdk.data_types import trace_tree
 
+    from litellm._logging import MessageContentFilter
+
     logger: Final = logging.getLogger(__name__)
+    logger.addFilter(MessageContentFilter())
 
     K = TypeVar("K", bound=str)
     V = TypeVar("V")
diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py
index 4dd33fa5565..c1ef3b30ed8 100644
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/litellm/litellm_core_utils/litellm_logging.py
@@ -6,6 +6,7 @@ import copy
 import datetime
 import functools
 import json
+import logging
 import math
 import os
 import re
@@ -26,7 +27,9 @@ from pydantic import BaseModel, JsonValue, TypeAdapter
 import litellm
 from litellm._internal_context import in_mirrored_call, outside_mirrored_call, post_response_phase
 from litellm._logging import (
+    diagnostic_message_logging_disabled,
     is_debugging_on,
+    log_safe_diagnostic,
     redact_string,
     session_id_var,
     set_session_id,
@@ -1724,6 +1727,13 @@ class Logging(LiteLLMLoggingBaseClass):
 
         Prints the RAW curl command sent from LiteLLM
         """
+        if diagnostic_message_logging_disabled():
+            log_safe_diagnostic(
+                verbose_logger,
+                "request_sent",
+                level=logging.WARNING if self.litellm_request_debug else logging.DEBUG,
+            )
+            return
         if is_debugging_on() or self.litellm_request_debug:
             if litellm.json_logs:
                 masked_headers: Final = self._get_masked_headers(headers or {})
@@ -1804,11 +1814,18 @@ class Logging(LiteLLMLoggingBaseClass):
                 attr = "debug"
 
             callattr: Final = verbose_logger.warning if attr == "warning" else verbose_logger.debug
-            callattr(
-                "RAW RESPONSE:\n{}\n\n".format(
-                    self._model_call_details_view.get("original_response", self._model_call_details_view)
+            if diagnostic_message_logging_disabled():
+                log_safe_diagnostic(
+                    verbose_logger,
+                    "response_received",
+                    level=logging.WARNING if self.litellm_request_debug else logging.DEBUG,
+                )
+            else:
+                callattr(
+                    "RAW RESPONSE:\n{}\n\n".format(
+                        self._model_call_details_view.get("original_response", self._model_call_details_view)
+                    )
                 )
-            )
             if getattr(self, "logger_fn", None) and callable(self.logger_fn):
                 try:
                     self.logger_fn(
@@ -7767,7 +7784,12 @@ def get_standard_logging_object_payload(
 def emit_standard_logging_payload(payload: StandardLoggingPayload):
     if os.getenv("LITELLM_PRINT_STANDARD_LOGGING_PAYLOAD"):
         try:
-            print(json.dumps(payload, indent=4, default=str), flush=True)  # noqa: T201
+            print(  # noqa: T201  # explicitly enabled diagnostic stdout
+                REDACTED_BY_LITELLM
+                if diagnostic_message_logging_disabled()
+                else json.dumps(payload, indent=4, default=str),
+                flush=True,
+            )
         except Exception as e:  # noqa: BLE001 # Safe catch-all for verbose logging
             verbose_logger.exception("Error serializing standard logging payload for debug output: %s", e)
 
diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py
index 5542249d2c7..5b26700bd33 100644
--- a/litellm/litellm_core_utils/streaming_handler.py
+++ b/litellm/litellm_core_utils/streaming_handler.py
@@ -163,10 +163,16 @@ def is_async_iterable(obj: object) -> bool:
     return isinstance(obj, collections.abc.AsyncIterable)
 
 
-def print_verbose(print_statement: object):
+def print_verbose(print_statement: object, *args: object):
     try:
         if litellm.set_verbose:
-            print(print_statement)  # noqa: T201
+            print(  # noqa: T201
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else str(print_statement) % args
+                if args
+                else print_statement
+            )
     except Exception:
         pass
 
@@ -666,10 +672,10 @@ class CustomStreamWrapper:
             text = ""
             is_finished = False
             finish_reason = ""
-            print_verbose(f"chunk: {chunk}")
+            print_verbose("chunk: %s", cast(object, chunk))  # cast-ok: logging accepts any object
             if chunk.startswith("data:"):
                 data_json: Final[_PredibaseStreamData] = json.loads(chunk[5:])
-                print_verbose(f"data json: {data_json}")
+                print_verbose("data json: %s", data_json)
                 if "token" in data_json and "text" in data_json["token"]:
                     text = data_json["token"]["text"]
                 if data_json.get("details", False) and data_json["details"].get("finish_reason", False):
@@ -859,7 +865,8 @@ class CustomStreamWrapper:
                         "completion_tokens": 0,
                     }
             else:
-                print_verbose(f"chunk: {chunk} (Type: {type(chunk)})")
+                if not litellm.redact_messages_in_debug_logs:
+                    print_verbose(f"chunk: {chunk} (Type: {type(chunk)})")
                 raise ValueError(f"Unable to parse response. Original response: {chunk}")
             text: Final = parsed_response.get("text_output", "")
             finish_reason: Final = parsed_response.get("stop_reason")
@@ -1500,13 +1507,21 @@ class CustomStreamWrapper:
         elif self.custom_llm_provider == "triton":
             response_obj = self.handle_triton_stream(chunk)
             completion_obj["content"] = response_obj["text"]
-            print_verbose(f"completion obj content: {completion_obj['content']}")
+            print_verbose(
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"completion obj content: {completion_obj['content']}"
+            )
             if response_obj["is_finished"]:
                 self.received_finish_reason = response_obj["finish_reason"]
         elif self.custom_llm_provider == "text-completion-openai":
             response_obj = self.handle_openai_text_completion_chunk(chunk)
             completion_obj["content"] = response_obj["text"]
-            print_verbose(f"completion obj content: {completion_obj['content']}")
+            print_verbose(
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"completion obj content: {completion_obj['content']}"
+            )
             if response_obj["is_finished"]:
                 self.received_finish_reason = response_obj["finish_reason"]
             if response_obj["usage"] is not None:
@@ -1528,7 +1543,11 @@ class CustomStreamWrapper:
                 litellm.CodestralTextCompletionConfig().chunk_parser(chunk),
             )
             completion_obj["content"] = response_obj["text"]
-            print_verbose(f"completion obj content: {completion_obj['content']}")
+            print_verbose(
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"completion obj content: {completion_obj['content']}"
+            )
             if response_obj["is_finished"]:
                 self.received_finish_reason = response_obj["finish_reason"]
             if "usage" in response_obj is not None:
@@ -1545,7 +1564,11 @@ class CustomStreamWrapper:
         elif self.custom_llm_provider == "azure_text":
             response_obj = self.handle_azure_text_completion_chunk(chunk)
             completion_obj["content"] = response_obj["text"]
-            print_verbose(f"completion obj content: {completion_obj['content']}")
+            print_verbose(
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"completion obj content: {completion_obj['content']}"
+            )
             if response_obj["is_finished"]:
                 self.received_finish_reason = response_obj["finish_reason"]
         elif self.custom_llm_provider == "cached_response":
@@ -1563,7 +1586,11 @@ class CustomStreamWrapper:
             completion_obj["content"] = response_obj["text"]
             if response_obj["tool_calls"] is not None:
                 completion_obj["tool_calls"] = response_obj["tool_calls"]
-            print_verbose(f"completion obj content: {completion_obj['content']}")
+            print_verbose(
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"completion obj content: {completion_obj['content']}"
+            )
             if hasattr(cached_chunk, "id"):
                 model_response.id = cached_chunk.id
                 self.response_id = cached_chunk.id
@@ -2083,11 +2110,13 @@ class CustomStreamWrapper:
                 else:
                     chunk = next(self.completion_stream)
                 if chunk is not None and chunk != b"":
-                    print_verbose(
-                        f"PROCESSED CHUNK PRE CHUNK CREATOR: {chunk.decode('utf-8', errors='replace') if isinstance(chunk, bytes) else chunk}; custom_llm_provider: {self.custom_llm_provider}"
-                    )
+                    if not litellm.redact_messages_in_debug_logs:
+                        print_verbose(
+                            f"PROCESSED CHUNK PRE CHUNK CREATOR: {chunk.decode('utf-8', errors='replace') if isinstance(chunk, bytes) else chunk}; custom_llm_provider: {self.custom_llm_provider}"
+                        )
                     response: ModelResponseStream | None = self.chunk_creator(chunk=chunk)
-                    print_verbose(f"PROCESSED CHUNK POST CHUNK CREATOR: {response}")
+                    if not litellm.redact_messages_in_debug_logs:
+                        print_verbose(f"PROCESSED CHUNK POST CHUNK CREATOR: {response}")
 
                     if response is None:
                         continue
diff --git a/litellm/llms/codestral/completion/handler.py b/litellm/llms/codestral/completion/handler.py
index 6b172e77295..5136f9a1e4e 100644
--- a/litellm/llms/codestral/completion/handler.py
+++ b/litellm/llms/codestral/completion/handler.py
@@ -182,7 +182,11 @@ class CodestralTextCompletion:
             original_response=response.text,
             additional_args={"complete_input_dict": data},
         )
-        print_verbose(f"codestral api: raw model_response: {response.text}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"codestral api: raw model_response: {response.text}"
+        )
         ## RESPONSE OBJECT
         if response.status_code != 200:
             raise TextCompletionCodestralError(
diff --git a/litellm/llms/deprecated_providers/aleph_alpha.py b/litellm/llms/deprecated_providers/aleph_alpha.py
index cb56717993e..65c5a14451f 100644
--- a/litellm/llms/deprecated_providers/aleph_alpha.py
+++ b/litellm/llms/deprecated_providers/aleph_alpha.py
@@ -247,7 +247,9 @@ def completion(
             original_response=response.text,
             additional_args={"complete_input_dict": data},
         )
-        print_verbose(f"raw model_response: {response.text}")
+        print_verbose(
+            "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"raw model_response: {response.text}"
+        )
         ## RESPONSE OBJECT
         completion_response: Final = response.json()
         if "error" in completion_response:
diff --git a/litellm/llms/huggingface/chat/transformation.py b/litellm/llms/huggingface/chat/transformation.py
index 3fcb738eacb..ed51f4cd90e 100644
--- a/litellm/llms/huggingface/chat/transformation.py
+++ b/litellm/llms/huggingface/chat/transformation.py
@@ -14,12 +14,14 @@ if TYPE_CHECKING:
 else:
     LoggingClass = Any
 
+from litellm._logging import MessageContentFilter
 from litellm.llms.base_llm.chat.transformation import BaseLLMException
 
 from ...openai.chat.gpt_transformation import OpenAIGPTConfig
 from ..common_utils import HuggingFaceError, fetch_inference_provider_mapping
 
 logger: Final = logging.getLogger(__name__)
+logger.addFilter(MessageContentFilter())
 
 BASE_URL: Final = "https://router.huggingface.co"
 
diff --git a/litellm/llms/replicate/chat/handler.py b/litellm/llms/replicate/chat/handler.py
index ad267847e81..abaff00f06c 100644
--- a/litellm/llms/replicate/chat/handler.py
+++ b/litellm/llms/replicate/chat/handler.py
@@ -52,7 +52,9 @@ def handle_prediction_response_streaming(
                         headers=response.headers,
                     )
                 new_output = output_string[len(previous_output) :]
-                print_verbose(f"New chunk: {new_output}")
+                print_verbose(
+                    "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"New chunk: {new_output}"
+                )
                 yield {"output": new_output, "status": status}
                 previous_output = output_string
             status = response_data["status"]
@@ -66,7 +68,9 @@ def handle_prediction_response_streaming(
         else:
             # this can fail temporarily but it does not mean the replicate request failed, replicate request fails when status=="failed"
             print_verbose(
-                f"Replicate: Failed to fetch prediction status and output.{response.status_code}{response.text}"
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"Replicate: Failed to fetch prediction status and output.{response.status_code}{response.text}"
             )
 
 
@@ -107,7 +111,9 @@ async def async_handle_prediction_response_streaming(
                         headers=response.headers,
                     )
                 new_output = output_string[len(previous_output) :]
-                print_verbose(f"New chunk: {new_output}")
+                print_verbose(
+                    "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"New chunk: {new_output}"
+                )
                 yield {"output": new_output, "status": status}
                 previous_output = output_string
             status = response_data.get("status", "")
@@ -121,7 +127,9 @@ async def async_handle_prediction_response_streaming(
         else:
             # this can fail temporarily but it does not mean the replicate request failed, replicate request fails when status=="failed"
             print_verbose(
-                f"Replicate: Failed to fetch prediction status and output.{response.status_code}{response.text}"
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"Replicate: Failed to fetch prediction status and output.{response.status_code}{response.text}"
             )
 
 
diff --git a/litellm/llms/sagemaker/completion/handler.py b/litellm/llms/sagemaker/completion/handler.py
index b89c59a0e5e..47928d02cee 100644
--- a/litellm/llms/sagemaker/completion/handler.py
+++ b/litellm/llms/sagemaker/completion/handler.py
@@ -635,7 +635,9 @@ class SagemakerLLM(BaseAWSLLM):
             additional_args={"complete_input_dict": data},
         )
 
-        print_verbose(f"raw model_response: {response}")
+        print_verbose(
+            "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"raw model_response: {response}"
+        )
 
         # Transform response based on model type
         from httpx import Response as HttpxResponse
diff --git a/litellm/llms/vllm/completion/handler.py b/litellm/llms/vllm/completion/handler.py
index f330f0928d0..5683a3c71c2 100644
--- a/litellm/llms/vllm/completion/handler.py
+++ b/litellm/llms/vllm/completion/handler.py
@@ -4,6 +4,7 @@ from typing import Final
 
 import httpx
 
+import litellm
 from litellm.litellm_core_utils.prompt_templates.factory import (
     prompt_factory,
     render_custom_prompt_template,
@@ -85,7 +86,9 @@ def completion(
             original_response=outputs,
             additional_args={"complete_input_dict": sampling_params},
         )
-        print_verbose(f"raw model_response: {outputs}")
+        print_verbose(
+            "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else f"raw model_response: {outputs}"
+        )
         ## RESPONSE OBJECT
         model_response.choices[0].message.content = outputs[0].outputs[0].text
 
diff --git a/litellm/main.py b/litellm/main.py
index d469aa629fe..f5a7d63ada1 100644
--- a/litellm/main.py
+++ b/litellm/main.py
@@ -946,11 +946,16 @@ def _text_completion_response_from_cached_dict(cached_response_dict: Mapping[str
 
 async def _async_streaming(response, model, custom_llm_provider, args):
     try:
-        print_verbose(f"received response in _async_streaming: {response}")
+        print_verbose(
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"received response in _async_streaming: {response}"
+        )
         if asyncio.iscoroutine(response):
             response = await response
         async for line in response:
-            print_verbose(f"line in async streaming: {line}")
+            if not litellm.redact_messages_in_debug_logs:
+                print_verbose(f"line in async streaming: {line}")
             yield line
     except Exception as e:
         custom_llm_provider = custom_llm_provider or "openai"
@@ -9403,7 +9408,7 @@ def print_verbose(print_statement: object):
     try:
         verbose_logger.debug(print_statement)
         if litellm.set_verbose:
-            print(print_statement)  # noqa: T201
+            print("redacted-by-litellm" if litellm.redact_messages_in_debug_logs else print_statement)  # noqa: T201
     except Exception:
         pass
 
diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py
index 7cd752c2297..519cfb52240 100644
--- a/litellm/proxy/common_request_processing.py
+++ b/litellm/proxy/common_request_processing.py
@@ -8,6 +8,7 @@ from dataclasses import dataclass
 from datetime import datetime, timezone
 from functools import lru_cache
 from itertools import chain
+from time import perf_counter
 from types import MappingProxyType
 from typing import (
     TYPE_CHECKING,
@@ -34,7 +35,7 @@ from starlette.types import Receive, Scope, Send
 from typing_extensions import Never
 
 import litellm
-from litellm._logging import redact_internal_details_from_client_message, verbose_proxy_logger
+from litellm._logging import log_safe_diagnostic, redact_internal_details_from_client_message, verbose_proxy_logger
 from litellm._uuid import uuid
 from litellm.anthropic_interface.exceptions import AnthropicErrorSseFrame, anthropic_error_sse_frame
 from litellm.constants import (
@@ -2877,7 +2878,7 @@ class ProxyBaseLLMRequestProcessing:
 
     def _debug_log_request_payload(self) -> None:
         """Log request payload at DEBUG level, truncating if too large."""
-        if not verbose_proxy_logger.isEnabledFor(logging.DEBUG):
+        if litellm.redact_messages_in_debug_logs or not verbose_proxy_logger.isEnabledFor(logging.DEBUG):
             return
         _payload_str: Final = json.dumps(self.data, default=str)
         if len(_payload_str) > MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG:
@@ -2933,7 +2934,12 @@ class ProxyBaseLLMRequestProcessing:
                 dispatched_call_type=self.route_type,
             )
 
-        return await open_sse_before_first_byte(
+        diagnostic_started: Final = (
+            perf_counter()
+            if litellm.redact_messages_in_debug_logs and verbose_proxy_logger.isEnabledFor(logging.DEBUG)
+            else None
+        )
+        response: Final = await open_sse_before_first_byte(
             self._process_llm_request(
                 request=request,
                 fastapi_response=fastapi_response,
@@ -2958,6 +2964,18 @@ class ProxyBaseLLMRequestProcessing:
             ping_interval_seconds=ttft_keepalive_interval(self.data, llm_router),
             on_late_failure=_audit_late_failure,
         )
+        if diagnostic_started is not None:
+            log_safe_diagnostic(
+                verbose_proxy_logger,
+                "response_ready",
+                status_code=(
+                    response.status_code
+                    if isinstance(response, Response)
+                    else fastapi_response.status_code or status.HTTP_200_OK
+                ),
+                elapsed_seconds=perf_counter() - diagnostic_started,
+            )
+        return response
 
     async def _process_llm_request(
         self,
@@ -4315,6 +4333,13 @@ class ProxyBaseLLMRequestProcessing:
         version: str | None = None,
     ) -> Never:
         """Raises ProxyException (OpenAI API compatible) if an exception is raised"""
+        if litellm.redact_messages_in_debug_logs:
+            log_safe_diagnostic(
+                verbose_proxy_logger,
+                "request_failed",
+                status_code=error_status_code(e, status.HTTP_500_INTERNAL_SERVER_ERROR),
+                level=logging.ERROR,
+            )
         log_llm_api_exception(e, self.litellm_call_id)
         billing_logging_obj: Final = _REQUEST_FIELDS.validate_python(self.data).get("litellm_logging_obj")
         # Allow callbacks to transform the error response
@@ -4601,6 +4626,7 @@ class ProxyBaseLLMRequestProcessing:
         cost_injection_enabled: Final = bool(getattr(litellm, "include_cost_in_streaming_usage", False))
         fast_path = not caps.has_streaming_chunk_override and not caps.has_guardrail and not cost_injection_enabled
         debug_enabled: Final = verbose_proxy_logger.isEnabledFor(logging.DEBUG)
+        diagnostic_started: Final = perf_counter() if litellm.redact_messages_in_debug_logs and debug_enabled else None
         stream_completed = False
         client_disconnected = False
         delivered_chunk = False
@@ -4619,7 +4645,7 @@ class ProxyBaseLLMRequestProcessing:
             async for chunk in guarded_stream:
                 # ``.format(chunk)`` was previously evaluated for every chunk
                 # regardless of log level; gate it behind the level check.
-                if debug_enabled:
+                if debug_enabled and not litellm.redact_messages_in_debug_logs:
                     verbose_proxy_logger.debug("async_data_generator: received streaming chunk - %s", chunk)
 
                 if not fast_path:
@@ -4662,6 +4688,11 @@ class ProxyBaseLLMRequestProcessing:
             if held_tail:
                 yield serialize_chunk(held_tail)
             stream_completed = True
+            log_safe_diagnostic(
+                verbose_proxy_logger,
+                "stream_completed",
+                elapsed_seconds=perf_counter() - diagnostic_started if diagnostic_started is not None else None,
+            )
         except (asyncio.CancelledError, GeneratorExit):
             # Client disconnected mid-stream. CancelledError / GeneratorExit
             # are BaseException and bypass the success/failure logging
@@ -4704,9 +4735,12 @@ class ProxyBaseLLMRequestProcessing:
                 e,
             )
 
+            stream_error_status: Final = error_status_code(e, status.HTTP_500_INTERNAL_SERVER_ERROR)
+            log_safe_diagnostic(
+                verbose_proxy_logger, "request_failed", status_code=stream_error_status, level=logging.ERROR
+            )
             if isinstance(e, HTTPException):
                 raise e
-            stream_error_status: Final = error_status_code(e, status.HTTP_500_INTERNAL_SERVER_ERROR)
             proxy_exception: Final = ProxyException(
                 message=redact_internal_details_from_client_message(getattr(e, "message", str(e))),
                 type=openai_error_type(e, stream_error_status),
diff --git a/litellm/proxy/common_utils/debug_utils.py b/litellm/proxy/common_utils/debug_utils.py
index 884aacb9fe1..cbdd87f1e7e 100644
--- a/litellm/proxy/common_utils/debug_utils.py
+++ b/litellm/proxy/common_utils/debug_utils.py
@@ -14,6 +14,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import TypeAdapter
 from typing_extensions import ReadOnly
 
+import litellm
 from litellm import get_secret_str
 from litellm._logging import verbose_proxy_logger
 from litellm.constants import PYTHON_GC_THRESHOLD
@@ -868,7 +869,7 @@ async def get_otel_spans():
     else:
         recorded_spans = []
 
-    print("Spans: ", recorded_spans)  # noqa: T201
+    print("Spans: ", "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else recorded_spans)  # noqa: T201
 
     most_recent_parent = None
     most_recent_start_time = 1000000
@@ -963,4 +964,7 @@ def init_verbose_loggers():
     except Exception as e:
         import logging
 
-        logging.warning("Failed to init verbose loggers: %s", e)
+        logging.warning(
+            "Failed to init verbose loggers: %s",
+            "redacted-by-litellm" if litellm.redact_messages_in_debug_logs else e,
+        )
diff --git a/litellm/proxy/guardrails/guardrail_hooks/presidio.py b/litellm/proxy/guardrails/guardrail_hooks/presidio.py
index 7e2f6bfec33..36841eb1221 100644
--- a/litellm/proxy/guardrails/guardrail_hooks/presidio.py
+++ b/litellm/proxy/guardrails/guardrail_hooks/presidio.py
@@ -1770,7 +1770,7 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
         try:
             verbose_proxy_logger.debug(print_statement)
             if litellm.set_verbose:
-                print(print_statement)  # noqa: T201
+                print("redacted-by-litellm" if litellm.redact_messages_in_debug_logs else print_statement)  # noqa: T201
         except Exception:
             pass
 
diff --git a/litellm/proxy/health_check.py b/litellm/proxy/health_check.py
index 73c2fc35492..b71b5436316 100644
--- a/litellm/proxy/health_check.py
+++ b/litellm/proxy/health_check.py
@@ -18,7 +18,10 @@ import litellm
 if TYPE_CHECKING:
     from litellm.router import Router
 
+from litellm._logging import MessageContentFilter
+
 logger: Final = logging.getLogger(__name__)
+logger.addFilter(MessageContentFilter())
 _DeploymentT: Final = TypeVar("_DeploymentT", bound=Mapping[str, object])
 from litellm.constants import (
     BACKGROUND_HEALTH_CHECK_MAX_TOKENS,
diff --git a/litellm/proxy/hooks/batch_redis_get.py b/litellm/proxy/hooks/batch_redis_get.py
index 94eb0bb52d1..da1ab24efc9 100644
--- a/litellm/proxy/hooks/batch_redis_get.py
+++ b/litellm/proxy/hooks/batch_redis_get.py
@@ -29,7 +29,7 @@ class _PROXY_BatchRedisRequests(CustomLogger):
         if debug_level == "DEBUG" or debug_level == "INFO":
             verbose_proxy_logger.debug(print_statement)
         if litellm.set_verbose is True:
-            print(print_statement)  # noqa: T201
+            print("redacted-by-litellm" if litellm.redact_messages_in_debug_logs else print_statement)  # noqa: T201
 
     async def async_pre_call_hook(
         self,
diff --git a/litellm/proxy/hooks/parallel_request_limiter.py b/litellm/proxy/hooks/parallel_request_limiter.py
index 05c64317a44..567f584803c 100644
--- a/litellm/proxy/hooks/parallel_request_limiter.py
+++ b/litellm/proxy/hooks/parallel_request_limiter.py
@@ -75,7 +75,7 @@ class _PROXY_MaxParallelRequestsHandler(CustomLogger):
         try:
             verbose_proxy_logger.debug(print_statement)
             if litellm.set_verbose:
-                print(print_statement)  # noqa: T201
+                print("redacted-by-litellm" if litellm.redact_messages_in_debug_logs else print_statement)  # noqa: T201
         except Exception:
             pass
 
diff --git a/litellm/proxy/hooks/prompt_injection_detection.py b/litellm/proxy/hooks/prompt_injection_detection.py
index e24ca3ce64e..30f87526730 100644
--- a/litellm/proxy/hooks/prompt_injection_detection.py
+++ b/litellm/proxy/hooks/prompt_injection_detection.py
@@ -83,7 +83,7 @@ class _OPTIONAL_PromptInjectionDetection(CustomLogger):
             verbose_proxy_logger.debug(print_statement)
 
         if litellm.set_verbose is True:
-            print(print_statement)  # noqa: T201
+            print("redacted-by-litellm" if litellm.redact_messages_in_debug_logs else print_statement)  # noqa: T201
 
     def update_environment(self, router: Router | None = None) -> None:
         self.llm_router = router
diff --git a/litellm/proxy/post_call_rules.py b/litellm/proxy/post_call_rules.py
index 6200bee4d7b..ebc87a3e3d1 100644
--- a/litellm/proxy/post_call_rules.py
+++ b/litellm/proxy/post_call_rules.py
@@ -1,5 +1,14 @@
-def post_response_rule(input):  # receives the model response
-    print(f"post_response_rule:input={input}")  # noqa: T201
+from collections.abc import Mapping
+
+import litellm
+
+
+def post_response_rule(input: str) -> Mapping[str, bool | str]:
+    print(  # noqa: T201  # legacy stdout diagnostic
+        "post_response_rule:input=redacted-by-litellm"
+        if litellm.redact_messages_in_debug_logs
+        else f"post_response_rule:input={input}"
+    )
     if len(input) < 200:
         return {
             "decision": False,
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index b0ccf74c63b..9acdae13f97 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -1163,13 +1163,23 @@ def _redact_worker_config_for_logging(worker_config: str | dict[str, JsonValue]
     (`LITELLM_DISABLE_REDACT_SECRETS=true`, an older log sink, a downstream
     handler that captures records pre-filter). Redact at the source.
     """
+    if litellm.redact_messages_in_debug_logs:
+        return "redacted-by-litellm"
     if worker_config is None:
         return None
-    if isinstance(worker_config, dict):
-        return _redact_secret_values_in_obj(worker_config)
-    parsed: Final[JsonValue] = safe_json_loads(worker_config, default=None)
+    parsed: Final[JsonValue] = (
+        worker_config if isinstance(worker_config, dict) else safe_json_loads(worker_config, default=None)
+    )
     if isinstance(parsed, dict):
-        return safe_dumps(_redact_secret_values_in_obj(parsed))
+        settings: Final = parsed.get("litellm_settings")
+        if isinstance(settings, dict) and settings.get("redact_messages_in_debug_logs") is not None:
+            try:
+                if TypeAdapter(bool).validate_python(settings["redact_messages_in_debug_logs"]):
+                    return "redacted-by-litellm"
+            except ValidationError:
+                return "redacted-by-litellm"
+        redacted: Final = _redact_secret_values_in_obj(parsed)
+        return redacted if isinstance(worker_config, dict) else safe_dumps(redacted)
     return worker_config
 
 
@@ -4990,27 +5000,45 @@ def sentry_is_configured(general_settings: Mapping[str, object]) -> bool:
     )
 
 
-def _scrub_guardrail_inner(inner: dict[str, JsonValue]) -> None:
+_DbConfigWarning: TypeAlias = tuple[str, tuple[object, ...]]
+
+
+def _warn_db_config(
+    message: str, *args: object, warning_sink: Callable[[_DbConfigWarning], None] | None = None
+) -> None:
+    if warning_sink is None:
+        verbose_proxy_logger.warning(message, *args)
+    else:
+        warning_sink((message, args))
+
+
+def _scrub_guardrail_inner(
+    inner: dict[str, JsonValue], warning_sink: Callable[[_DbConfigWarning], None] | None = None
+) -> None:
     """Strip remote-URL entries from a guardrail's ``callbacks`` list
     and ``guardrail`` (v2 module-path) field. Mutates in place."""
     cbs: Final = inner.get("callbacks")
     if isinstance(cbs, list):
         cleaned: Final = [c for c in cbs if not _is_remote_module_url(c)]
         if len(cleaned) != len(cbs):
-            verbose_proxy_logger.warning(
+            _warn_db_config(
                 "Refused %d remote-URL entries from DB-overlay litellm_settings.guardrails[...].callbacks",
                 len(cbs) - len(cleaned),
+                warning_sink=warning_sink,
             )
             inner["callbacks"] = cleaned
     if _is_remote_module_url(inner.get("guardrail")):
-        verbose_proxy_logger.warning(
+        _warn_db_config(
             "Refused remote-URL guardrail module from DB-overlay litellm_settings.guardrails[...].guardrail: %r",
             inner.get("guardrail"),
+            warning_sink=warning_sink,
         )
         inner["guardrail"] = None
 
 
-def _scrub_db_overlay_remote_module_loads(section: str, db_value: object) -> object:
+def _scrub_db_overlay_remote_module_loads(
+    section: str, db_value: object, warning_sink: Callable[[_DbConfigWarning], None] | None = None
+) -> object:
     """Strip ``s3://`` / ``gcs://`` entries from the DB-overlay value for
     fields whose contents reach ``get_instance_fn``. The same scheme is
     allowed from a YAML config (the documented operator flow) but a
@@ -5026,12 +5054,13 @@ def _scrub_db_overlay_remote_module_loads(section: str, db_value: object) -> obj
     for field in str_fields:
         v = sanitized.get(field)
         if _is_remote_module_url(v):
-            verbose_proxy_logger.warning(
+            _warn_db_config(
                 "Refused remote-URL value for DB-overlay %s.%s=%r; only "
                 "config.yaml entries may reference s3:// / gcs:// modules.",
                 section,
                 field,
                 v,
+                warning_sink=warning_sink,
             )
             sanitized[field] = None
     for field in list_fields:
@@ -5039,13 +5068,14 @@ def _scrub_db_overlay_remote_module_loads(section: str, db_value: object) -> obj
         if isinstance(v, list):
             cleaned = [item for item in v if not _is_remote_module_url(item)]
             if len(cleaned) != len(v):
-                verbose_proxy_logger.warning(
+                _warn_db_config(
                     "Refused %d remote-URL entries from DB-overlay %s.%s; "
                     "only config.yaml entries may reference s3:// / gcs:// "
                     "modules.",
                     len(v) - len(cleaned),
                     section,
                     field,
+                    warning_sink=warning_sink,
                 )
                 sanitized[field] = cleaned
     # ``custom_provider_map`` is a list of dicts with ``custom_handler`` —
@@ -5055,9 +5085,10 @@ def _scrub_db_overlay_remote_module_loads(section: str, db_value: object) -> obj
         if isinstance(cpm, list):
             for item in cpm:
                 if isinstance(item, dict) and _is_remote_module_url(item.get("custom_handler")):
-                    verbose_proxy_logger.warning(
+                    _warn_db_config(
                         "Refused remote-URL custom_handler from DB-overlay litellm_settings.custom_provider_map: %r",
                         item.get("custom_handler"),
+                        warning_sink=warning_sink,
                     )
                     item["custom_handler"] = None
     # ``litellm_settings.guardrails`` is a list of single-key dicts in
@@ -5075,19 +5106,20 @@ def _scrub_db_overlay_remote_module_loads(section: str, db_value: object) -> obj
                 for inner in entry.values():
                     if not isinstance(inner, dict):
                         continue
-                    _scrub_guardrail_inner(inner)
+                    _scrub_guardrail_inner(inner, warning_sink=warning_sink)
                 lp = entry.get("litellm_params")
                 if isinstance(lp, dict):
-                    _scrub_guardrail_inner(lp)
+                    _scrub_guardrail_inner(lp, warning_sink=warning_sink)
 
     # ``general_settings.litellm_jwtauth.custom_validate`` is a nested
     # string field.
     if section == "general_settings":
         jwt: Final = sanitized.get("litellm_jwtauth")
         if isinstance(jwt, dict) and _is_remote_module_url(jwt.get("custom_validate")):
-            verbose_proxy_logger.warning(
+            _warn_db_config(
                 "Refused remote-URL custom_validate from DB-overlay general_settings.litellm_jwtauth: %r",
                 jwt.get("custom_validate"),
+                warning_sink=warning_sink,
             )
             jwt["custom_validate"] = None
         # ``pass_through_endpoints`` is a list of dicts whose ``target``
@@ -5099,12 +5131,13 @@ def _scrub_db_overlay_remote_module_loads(section: str, db_value: object) -> obj
         if isinstance(pte, list):
             for entry in pte:
                 if isinstance(entry, dict) and _is_remote_module_url(entry.get("target")):
-                    verbose_proxy_logger.warning(
+                    _warn_db_config(
                         "Refused remote-URL target from DB-overlay "
                         "general_settings.pass_through_endpoints "
                         "(path=%r): %r",
                         entry.get("path"),
                         entry.get("target"),
+                        warning_sink=warning_sink,
                     )
                     entry["target"] = None
     return sanitized
@@ -5794,6 +5827,11 @@ class ProxyConfig:
         }
 
     def _apply_resolved_runtime_settings(self, config: Mapping[str, object]) -> None:
+        diagnostic_settings: Final = config.get("litellm_settings")
+        if isinstance(diagnostic_settings, Mapping) and "redact_messages_in_debug_logs" in diagnostic_settings:
+            litellm.redact_messages_in_debug_logs = TypeAdapter(bool).validate_python(
+                diagnostic_settings["redact_messages_in_debug_logs"]
+            )
         for section, store in self._settings_stores.items():
             if isinstance(config.get(section), Mapping):
                 store.apply_runtime_values(_as_settings_mapping(config[section]))
@@ -6403,12 +6441,16 @@ class ProxyConfig:
 
         self._load_yaml_settings_stores(config)
 
+        pending_db_diagnostics: Final[list[tuple[str, JsonValue]]] = []  # mutable-ok: defer rows until privacy resolves
+        pending_db_warnings: Final[list[_DbConfigWarning]] = []  # mutable-ok: defer warnings until privacy resolves
         ## UPDATE CONFIG WITH DB
         if prisma_client is not None and store_model_in_db is True:
             config = await self._update_config_from_db(
                 config=config,
                 prisma_client=prisma_client,
                 store_model_in_db=store_model_in_db,
+                diagnostic_sink=pending_db_diagnostics.append,
+                warning_sink=pending_db_warnings.append,
             )
 
         config = self._config_with_resolved_settings(config)
@@ -6421,6 +6463,12 @@ class ProxyConfig:
 
         config = self._check_for_os_environ_vars(config=config)
         self._apply_resolved_runtime_settings(config)
+        for section, param_value in pending_db_diagnostics:
+            _log_db_config_param(section, param_value)
+        for message, args in pending_db_warnings:
+            verbose_proxy_logger.warning(message, *args)
+        if prisma_client is not None and store_model_in_db is True:
+            self._warn_about_shadowed_db_settings()
 
         self.update_config_state(config=config)
 
@@ -6973,6 +7021,8 @@ class ProxyConfig:
 
                     parse_budget_reset_time(value)
                     setattr(litellm, key, value)
+                elif key == "redact_messages_in_debug_logs":
+                    litellm.redact_messages_in_debug_logs = TypeAdapter(bool).validate_python(value)
                 elif key == "drop_params":
                     litellm.drop_params = drop_params_flag(value, "litellm_settings.drop_params", verbose_proxy_logger)
                 elif key == "suppress_startup_output":
@@ -8462,6 +8512,8 @@ class ProxyConfig:
         prisma_client: PrismaClient,
         config: dict,
         store_model_in_db: bool | None,
+        diagnostic_sink: Callable[[tuple[str, JsonValue]], None] | None = None,
+        warning_sink: Callable[[_DbConfigWarning], None] | None = None,
     ):
         if store_model_in_db is not True:
             verbose_proxy_logger.info("'store_model_in_db' is not True, skipping db updates")
@@ -8479,13 +8531,14 @@ class ProxyConfig:
             if param_value is None:
                 continue
 
-            verbose_proxy_logger.debug(
-                "param_name=%s, param_value=%s",
-                section,
-                _redact_config_param_value_for_logging(section, param_value),
-            )
+            if diagnostic_sink is None:
+                _log_db_config_param(section, param_value)
+            else:
+                diagnostic_sink((section, param_value))
             if section == "litellm_settings":
-                self._apply_litellm_settings_db_values(self._prepared_db_settings_values(section, param_value))
+                self._apply_litellm_settings_db_values(
+                    self._prepared_db_settings_values(section, param_value, warning_sink=warning_sink)
+                )
             elif section == "environment_variables":
                 await self._apply_db_environment_variables_row(
                     prisma_client, response, environment_before_db, environment_read
@@ -8493,10 +8546,11 @@ class ProxyConfig:
             else:
                 self._settings_stores[section].apply_db_row(
                     section,
-                    self._prepared_db_settings_values(section, param_value),
+                    self._prepared_db_settings_values(section, param_value, warning_sink=warning_sink),
                 )
 
-        self._warn_about_shadowed_db_settings()
+        if warning_sink is None:
+            self._warn_about_shadowed_db_settings()
         return self._config_with_resolved_settings(config)
 
     async def _apply_db_environment_variables_row(
@@ -8551,12 +8605,17 @@ class ProxyConfig:
         )
         for section, key in sorted(shadowed - self._warned_shadowed_keys):
             verbose_proxy_logger.warning(
-                "%s", config_ownership_message(section=section, key=key, shadows_db_value=True)
+                "%s",
+                config_ownership_message(section=section, key=key, shadows_db_value=True),
             )
         self._warned_shadowed_keys = shadowed
 
-    def _prepared_db_settings_values(self, section: Section, value: object) -> Mapping[str, SettingsJsonValue]:
-        scrubbed: Final = _scrub_db_overlay_remote_module_loads(section=section, db_value=value)
+    def _prepared_db_settings_values(
+        self, section: Section, value: object, warning_sink: Callable[[_DbConfigWarning], None] | None = None
+    ) -> Mapping[str, SettingsJsonValue]:
+        scrubbed: Final = _scrub_db_overlay_remote_module_loads(
+            section=section, db_value=value, warning_sink=warning_sink
+        )
         return _as_settings_mapping(scrubbed)
 
     def _apply_db_environment_variables(
@@ -20139,6 +20198,14 @@ def _redact_config_param_value_for_logging(param_name: str | None, param_value:
     return param_value
 
 
+def _log_db_config_param(section: str, param_value: JsonValue) -> None:
+    verbose_proxy_logger.debug(
+        "param_name=%s, param_value=%s",
+        section,
+        _redact_config_param_value_for_logging(section, param_value),
+    )
+
+
 def _redact_general_setting_value(field_name: str, value: JsonValue, is_full_admin: bool) -> JsonValue:
     if is_full_admin:
         return value
diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py
index 9e125b95a68..9f4de03ae16 100644
--- a/litellm/proxy/utils.py
+++ b/litellm/proxy/utils.py
@@ -435,9 +435,15 @@ def print_verbose(print_statement: object):
     """
     import traceback
 
-    verbose_proxy_logger.debug("%s\n%s", print_statement, traceback.format_exc())
+    verbose_proxy_logger.debug(
+        "%s\n%s", print_statement, "" if litellm.redact_messages_in_debug_logs else traceback.format_exc()
+    )
     if litellm.set_verbose:
-        print(f"LiteLLM Proxy: {redact_string(str(print_statement))}")  # noqa: T201
+        print(  # noqa: T201  # legacy stdout diagnostic
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"LiteLLM Proxy: {redact_string(str(print_statement))}"
+        )
 
 
 def _get_email_logger_class():
diff --git a/litellm/responses/mcp/mcp_streaming_iterator.py b/litellm/responses/mcp/mcp_streaming_iterator.py
index 189366feab4..0b5c8a8fbb6 100644
--- a/litellm/responses/mcp/mcp_streaming_iterator.py
+++ b/litellm/responses/mcp/mcp_streaming_iterator.py
@@ -1,6 +1,7 @@
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Final, cast
 
+import litellm
 from litellm._logging import verbose_logger
 from litellm._uuid import uuid
 from litellm.litellm_core_utils.agentic_loop_settings import validated_max_agentic_loops
@@ -150,7 +151,8 @@ async def create_mcp_list_tools_events(
         verbose_logger.error("Error creating MCP list tools events: %s", e)
         import traceback
 
-        traceback.print_exc()
+        if not litellm.redact_messages_in_debug_logs:
+            traceback.print_exc()
 
         # Emit failed event on error
         failed_event: Final = MCPListToolsFailedEvent(
@@ -816,7 +818,8 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
             verbose_logger.error("Error creating initial response iterator: %s", e)
             import traceback
 
-            traceback.print_exc()
+            if not litellm.redact_messages_in_debug_logs:
+                traceback.print_exc()
             self.base_iterator = None
             self._initial_creation_error = e
             self._stream_error = e
@@ -976,7 +979,8 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
             verbose_logger.error("Error in tool execution: %s", e)
             import traceback
 
-            traceback.print_exc()
+            if not litellm.redact_messages_in_debug_logs:
+                traceback.print_exc()
             self.tool_results = []
             self._tool_results_for_response = self.collected_response
 
@@ -1052,7 +1056,8 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
             verbose_logger.error("Error creating follow-up iterator: %s", e)
             import traceback
 
-            traceback.print_exc()
+            if not litellm.redact_messages_in_debug_logs:
+                traceback.print_exc()
             self.base_iterator = None
             self._stream_error = e
 
diff --git a/litellm/router.py b/litellm/router.py
index 3a423b79e49..195f38c5edb 100644
--- a/litellm/router.py
+++ b/litellm/router.py
@@ -75,7 +75,7 @@ import litellm
 import litellm.litellm_core_utils.exception_mapping_utils
 from litellm import get_secret_str
 from litellm._internal_context import in_mirrored_call, mirrored_call, service_target, with_service_target
-from litellm._logging import verbose_router_logger
+from litellm._logging import log_safe_diagnostic, verbose_router_logger
 from litellm._uuid import uuid
 from litellm.caching.caching import (
     DualCache,
@@ -7111,6 +7111,8 @@ class Router(DeprecatedRouterMembers):
                 trigger, healthy_deployments, all_deployments, initial_kwargs
             ):
                 return _StreamRetriesExhausted(last_error)
+            if litellm.redact_messages_in_debug_logs:
+                log_safe_diagnostic(verbose_router_logger, "retry", attempt=attempt + 1)
             if get_metadata_variable_name_from_kwargs(retry_kwargs) in retry_kwargs:
                 self._core_log_retry(kwargs=retry_kwargs, e=trigger)
             await asyncio.sleep(
@@ -7965,7 +7967,8 @@ class Router(DeprecatedRouterMembers):
                 except Exception as e:
                     import traceback
 
-                    traceback.print_exc()
+                    if not litellm.redact_messages_in_debug_logs:
+                        traceback.print_exc()
                     receieved_exceptions.append(e)
                     return None
 
@@ -9412,6 +9415,8 @@ class Router(DeprecatedRouterMembers):
             await asyncio.sleep(retry_after)
 
             for current_attempt in range(num_retries):
+                if litellm.redact_messages_in_debug_logs:
+                    log_safe_diagnostic(verbose_router_logger, "retry", attempt=current_attempt + 1)
                 try:
                     # Update retry tracking metadata before each retry attempt
                     _metadata["attempted_retries"] = current_attempt + 1
diff --git a/litellm/router_utils/fallback_event_handlers.py b/litellm/router_utils/fallback_event_handlers.py
index 6d1de1f9f6c..6875ece8e29 100644
--- a/litellm/router_utils/fallback_event_handlers.py
+++ b/litellm/router_utils/fallback_event_handlers.py
@@ -1,5 +1,6 @@
 import hashlib
 import json
+import logging
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -11,7 +12,7 @@ import httpx
 from pydantic import TypeAdapter
 
 import litellm
-from litellm._logging import verbose_router_logger
+from litellm._logging import log_safe_diagnostic, verbose_router_logger
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.batch_utils import get_router_metadata_variable_name
 from litellm.litellm_core_utils.core_helpers import get_metadata_variable_name_from_kwargs, safe_deep_copy
@@ -873,7 +874,10 @@ async def run_async_fallback(
         try:
             # LOGGING
             kwargs = litellm_router.log_retry(kwargs=kwargs, e=original_exception)
-            verbose_router_logger.info("Falling back to model_group = %s", mask_sensitive_structure(mg))
+            if litellm.redact_messages_in_debug_logs:
+                log_safe_diagnostic(verbose_router_logger, "fallback", attempt=fallback_depth + 1, level=logging.INFO)
+            else:
+                verbose_router_logger.info("Falling back to model_group = %s", mask_sensitive_structure(mg))
             kwargs.pop("_target_order", None)
             if isinstance(mg, str):
                 kwargs["model"] = mg
diff --git a/litellm/rust_bridge/logger.py b/litellm/rust_bridge/logger.py
index 544e7195848..37ba5b6eab0 100644
--- a/litellm/rust_bridge/logger.py
+++ b/litellm/rust_bridge/logger.py
@@ -5,6 +5,7 @@ from typing import Final
 
 from pydantic import JsonValue
 
+import litellm
 from litellm._logging import (
     CorrelationContextFilter,
     DiagnosticProcessingFilter,
@@ -24,7 +25,7 @@ def context() -> tuple[str, str]:
 
 
 def enabled(level: int) -> bool:
-    return verbose_logger.isEnabledFor(level)
+    return not getattr(litellm, "redact_messages_in_debug_logs", False) and verbose_logger.isEnabledFor(level)
 
 
 def emit(
diff --git a/litellm/utils.py b/litellm/utils.py
index 549590e1cf5..4207b5c8c1c 100644
--- a/litellm/utils.py
+++ b/litellm/utils.py
@@ -587,7 +587,7 @@ def print_verbose(
         elif log_level == "ERROR":
             verbose_logger.error(print_statement)
         if litellm.set_verbose is True and logger_only is False:
-            print(print_statement)  # noqa: T201
+            print("redacted-by-litellm" if litellm.redact_messages_in_debug_logs else print_statement)  # noqa: T201
     except Exception:
         pass
 
@@ -2069,7 +2069,9 @@ def client(original_function):
 
             # [OPTIONAL] CHECK CACHE
             print_verbose(
-                f"SYNC kwargs[caching]: {kwargs.get('caching', False)}; litellm.cache: {litellm.cache}; kwargs.get('cache')['no-cache']: {kwargs.get('cache', {}).get('no-cache', False)}"
+                "redacted-by-litellm"
+                if litellm.redact_messages_in_debug_logs
+                else f"SYNC kwargs[caching]: {kwargs.get('caching', False)}; litellm.cache: {litellm.cache}; kwargs.get('cache')['no-cache']: {kwargs.get('cache', {}).get('no-cache', False)}"
             )
             # if caching is false or cache["no-cache"]==True, don't run this
             if (
@@ -2352,7 +2354,9 @@ def client(original_function):
             # [OPTIONAL] CHECK CACHE
             if is_debugging_on():
                 print_verbose(
-                    f"ASYNC kwargs[caching]: {kwargs.get('caching', False)}; litellm.cache: {litellm.cache}; kwargs.get('cache'): {kwargs.get('cache', None)}"
+                    "redacted-by-litellm"
+                    if litellm.redact_messages_in_debug_logs
+                    else f"ASYNC kwargs[caching]: {kwargs.get('caching', False)}; litellm.cache: {litellm.cache}; kwargs.get('cache'): {kwargs.get('cache', None)}"
                 )
             _caching_handler_response: CachingHandlerResponse | None = await _llm_caching_handler.async_get_cache(
                 model=model or "",
@@ -8618,7 +8622,9 @@ def trim_messages(
 
         #### Trimming messages if current_tokens > max_tokens
         print_verbose(
-            f"Need to trim input messages: {messages}, current_tokens{current_tokens}, max_tokens: {max_tokens}"
+            "redacted-by-litellm"
+            if litellm.redact_messages_in_debug_logs
+            else f"Need to trim input messages: {messages}, current_tokens{current_tokens}, max_tokens: {max_tokens}"
         )
         system_message_event: dict | None = None
         if system_message:
@@ -8878,6 +8884,9 @@ def _debug_params_repr(params: Mapping[str, object]) -> str:
 def print_args_passed_to_litellm(original_function, args, kwargs):
     if not is_debugging_on():
         return
+    if litellm.redact_messages_in_debug_logs:
+        print_verbose("redacted-by-litellm")
+        return
     try:
         # we've already printed this for acompletion, don't print for completion
         if (
diff --git a/tests/unit/integrations/cloudzero/test_cz_stream_api.py b/tests/unit/integrations/cloudzero/test_cz_stream_api.py
index d4e49a1252f..326daa3b368 100644
--- a/tests/unit/integrations/cloudzero/test_cz_stream_api.py
+++ b/tests/unit/integrations/cloudzero/test_cz_stream_api.py
@@ -1,12 +1,15 @@
+import json
 import zoneinfo
 from datetime import datetime, timezone
+from typing import Final
 from unittest.mock import MagicMock, Mock, patch
 
 import httpx
 import polars as pl
 import pytest
+import respx
 
-
+import litellm
 from litellm.integrations.cloudzero.cz_stream_api import CloudZeroStreamer
 
 
@@ -177,3 +180,64 @@ class TestCloudZeroStreamer:
         # Should avoid scientific notation
         assert "e" not in result["cost"].lower()
         assert "e" not in result["large_float"].lower()
+
+
+@pytest.mark.parametrize("redact", (False, True))
+def test_cloudzero_http_error_diagnostics_respect_privacy_without_changing_exception(
+    redact: bool, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    marker: Final = "synthetic-private-upstream-content"
+    streamer: Final = CloudZeroStreamer("test-key", "test-connection")
+    with respx.mock:
+        respx.post(f"{streamer.base_url}/v2/connections/billing/anycost/test-connection/billing_drops").respond(
+            status_code=400, text=marker
+        )
+        with pytest.raises(httpx.HTTPStatusError) as captured:
+            streamer._send_daily_batch("2026-01-01", pl.DataFrame({"cost/cost": [1.0]}), "replace_hourly")
+    assert captured.value.response.text == marker
+    assert (marker in capsys.readouterr().out) is not redact
+
+
+@pytest.mark.parametrize("redact", (False, True))
+def test_cloudzero_skips_invalid_timestamp_without_logging_content_when_private(
+    redact: bool, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    marker: Final = "synthetic-private-invalid-timestamp"
+    streamer: Final = CloudZeroStreamer("test-key", "test-connection")
+    data: Final = pl.DataFrame(
+        {"time/usage_start": [marker, "2026-01-01T00:00:00Z"], "cost/cost": [9.0, 1.0]}
+    )
+    with respx.mock:
+        route: Final = respx.post(
+            f"{streamer.base_url}/v2/connections/billing/anycost/test-connection/billing_drops"
+        ).respond(status_code=200)
+        streamer.send_batched(data)
+    assert len(route.calls) == 1
+    assert json.loads(route.calls[0].request.content) == {
+        "month": "2026-01",
+        "operation": "replace_hourly",
+        "data": [{"time/usage_start": "2026-01-01T00:00:00Z", "cost/cost": "1"}],
+    }
+    assert (marker in capsys.readouterr().out) is not redact
+
+
+@pytest.mark.parametrize("redact", (False, True))
+def test_cloudzero_network_error_preserved_without_logging_content_when_private(
+    redact: bool, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    marker: Final = "synthetic-private-network-error"
+    streamer: Final = CloudZeroStreamer("test-key", "test-connection")
+    failure: Final = httpx.ConnectError(marker)
+    with respx.mock:
+        respx.post(
+            f"{streamer.base_url}/v2/connections/billing/anycost/test-connection/billing_drops"
+        ).mock(side_effect=failure)
+        with pytest.raises(httpx.ConnectError) as captured:
+            streamer.send_batched(
+                pl.DataFrame({"time/usage_start": ["2026-01-01T00:00:00Z"], "cost/cost": [1.0]})
+            )
+    assert captured.value is failure
+    assert (marker in capsys.readouterr().out) is not redact
diff --git a/tests/unit/integrations/test_deepeval.py b/tests/unit/integrations/test_deepeval.py
index 6dd3aef0a03..09cb2873ff1 100644
--- a/tests/unit/integrations/test_deepeval.py
+++ b/tests/unit/integrations/test_deepeval.py
@@ -1,5 +1,10 @@
 import os
 import unittest
+from typing import Final
+
+import pytest
+
+import litellm
 from litellm._uuid import uuid
 from datetime import datetime, timezone
 from unittest.mock import MagicMock, patch
@@ -118,3 +123,22 @@ class TestDeepEvalLogger(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+@pytest.mark.parametrize("redact", (False, True))
+@pytest.mark.parametrize("exception_kind", ("missing", "none", "error"))
+def test_retry_diagnostic_hides_exception_and_dynamic_details_only_when_opted_in(
+    redact: bool, exception_kind: str, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from litellm.integrations.deepeval.api import log_retry_error
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    marker: Final = "synthetic-private-retry-content"
+    failure: Final = ValueError(marker) if exception_kind == "error" else None
+    details: Final = {"tries": 2} if exception_kind == "missing" else {"exception": failure, "tries": 2}
+    log_retry_error(details)
+    assert ("Confident AI Error:" in caplog.text) is (failure is not None)
+    assert (marker in caplog.text) is (failure is not None and not redact)
+    assert ("redacted-by-litellm" in caplog.text) is redact
+    assert details.get("exception") is failure
+    assert ("exception" in details) is (exception_kind != "missing")
diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py
index c320cb36d73..630818b2d15 100644
--- a/tests/unit/litellm_core_utils/test_litellm_logging.py
+++ b/tests/unit/litellm_core_utils/test_litellm_logging.py
@@ -13993,6 +13993,59 @@ def test_record_native_bedrock_guardrail_bills_nothing_for_a_response_served_fro
     assert "standard_logging_guardrail_information" not in logging_obj.model_call_details["litellm_params"]["metadata"]
 
 
+class _UnrenderableRequestBody:
+    def __repr__(self) -> str:
+        raise AssertionError("request body was formatted for diagnostics")
+
+
+@pytest.mark.parametrize("json_logs", [False, True])
+def test_private_debug_request_skips_body_formatting_and_keeps_transform_data(
+    logging_obj: LitellmLogging, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, json_logs: bool
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    monkeypatch.setattr(litellm, "json_logs", json_logs)
+    logging_obj.litellm_request_debug = True
+    body: Final = {"messages": _UnrenderableRequestBody()}
+    details: Final = {"api_base": "https://example.com", "complete_input_dict": body, "request_str": "PRIVATE_REQUEST"}
+    logging_obj._print_llm_call_debugging_log(api_base="https://example.com", headers={}, additional_args=details)
+    assert "event=request_sent" in caplog.text
+    assert "PRIVATE_REQUEST" not in caplog.text
+    assert details["complete_input_dict"] is body
+
+
+@pytest.mark.parametrize("diagnostic_privacy", [False, True])
+@pytest.mark.parametrize("callback_privacy", [False, True])
+def test_debug_privacy_preserves_raw_transform_request_body(
+    logging_obj: LitellmLogging, monkeypatch: pytest.MonkeyPatch, diagnostic_privacy: bool, callback_privacy: bool
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", diagnostic_privacy)
+    monkeypatch.setattr(litellm, "turn_off_message_logging", callback_privacy)
+    metadata: Final = _pre_call_with_raw_request_logging(logging_obj)
+    if callback_privacy:
+        assert metadata["raw_request"] == REDACTED_BY_LITELLM
+    else:
+        assert "PRIVATE" in metadata["raw_request"]
+    assert logging_obj.model_call_details["raw_request_typed_dict"]["raw_request_body"] == _PRIVATE_RAW_REQUEST_ARGS["complete_input_dict"]
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+def test_standard_payload_debug_stdout_obeys_diagnostic_privacy(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], enabled: bool
+) -> None:
+    from litellm.litellm_core_utils.litellm_logging import emit_standard_logging_payload
+
+    monkeypatch.setenv("LITELLM_PRINT_STANDARD_LOGGING_PAYLOAD", "1")
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", enabled)
+    payload: Final = StandardLoggingPayload(messages=[{"role": "user", "content": "PRIVATE_PAYLOAD"}])
+    emit_standard_logging_payload(payload)
+    output: Final = capsys.readouterr().out
+    if enabled:
+        assert output.strip() == REDACTED_BY_LITELLM
+    else:
+        assert json.loads(output)["messages"] == payload["messages"]
+    assert payload["messages"] == [{"role": "user", "content": "PRIVATE_PAYLOAD"}]
+
+
 def test_record_native_bedrock_guardrail_bills_each_call_its_own_trace_when_calls_share_one_metadata_dict() -> None:
     import datetime
 
diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py
index f4502937fce..8c53b8cbbf9 100644
--- a/tests/unit/litellm_core_utils/test_streaming_handler.py
+++ b/tests/unit/litellm_core_utils/test_streaming_handler.py
@@ -3758,11 +3758,17 @@ def test_dispatch_petals_empty_stream_after_finish_raises(
         _run_dispatch(initialized_custom_stream_wrapper, chunk=None)
 
 
+@pytest.mark.parametrize("redact", (False, True))
 def test_dispatch_cached_response_extracts_delta(
     initialized_custom_stream_wrapper: CustomStreamWrapper,
+    redact: bool,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
 ):
     """cached_response replays a stored ModelResponseStream; the dispatch lifts
     its delta content, finish_reason and id back onto the live response."""
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    monkeypatch.setattr(litellm, "set_verbose", True)
     initialized_custom_stream_wrapper.custom_llm_provider = "cached_response"
     chunk = ModelResponseStream(
         id="chatcmpl-cache-1",
@@ -3784,6 +3790,7 @@ def test_dispatch_cached_response_extracts_delta(
     assert initialized_custom_stream_wrapper.received_finish_reason == "stop"
     assert model_response.id == "chatcmpl-cache-1"
     assert initialized_custom_stream_wrapper.response_id == "chatcmpl-cache-1"
+    assert ("cached text" in capsys.readouterr().out) is not redact
 
 
 def test_dispatch_cached_response_without_choices_is_an_empty_chunk(
@@ -3955,11 +3962,17 @@ def test_dispatch_custom_provider_finish_only_returns_none_early(
     assert initialized_custom_stream_wrapper.received_finish_reason == "stop"
 
 
+@pytest.mark.parametrize("redact", (False, True))
 def test_dispatch_text_completion_codestral_parses_chunk(
     initialized_custom_stream_wrapper: CustomStreamWrapper,
+    redact: bool,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
 ):
     """text-completion-codestral streams raw SSE JSON strings that the dispatch
     routes through CodestralTextCompletionConfig to extract content/finish."""
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    monkeypatch.setattr(litellm, "set_verbose", True)
     initialized_custom_stream_wrapper.custom_llm_provider = "text-completion-codestral"
     chunk = json.dumps(
         {"choices": [{"delta": {"content": "codestral text"}, "finish_reason": "stop"}]}
@@ -3970,6 +3983,7 @@ def test_dispatch_text_completion_codestral_parses_chunk(
     assert isinstance(result, _ProviderChunkParsed)
     assert completion_obj["content"] == "codestral text"
     assert initialized_custom_stream_wrapper.received_finish_reason == "stop"
+    assert ("codestral text" in capsys.readouterr().out) is not redact
 
 
 def test_dispatch_text_completion_codestral_requires_string(
@@ -3983,10 +3997,16 @@ def test_dispatch_text_completion_codestral_requires_string(
         _run_dispatch(initialized_custom_stream_wrapper, {"not": "a string"})
 
 
+@pytest.mark.parametrize("redact", (False, True))
 def test_dispatch_triton_stream(
     initialized_custom_stream_wrapper: CustomStreamWrapper,
+    redact: bool,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
 ):
     """triton stream chunks arrive as dicts keyed by text_output/stop_reason."""
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    monkeypatch.setattr(litellm, "set_verbose", True)
     initialized_custom_stream_wrapper.custom_llm_provider = "triton"
     chunk = {"text_output": "triton text", "is_finished": True, "stop_reason": "stop"}
 
@@ -3995,13 +4015,20 @@ def test_dispatch_triton_stream(
     assert isinstance(result, _ProviderChunkParsed)
     assert completion_obj["content"] == "triton text"
     assert initialized_custom_stream_wrapper.received_finish_reason == "stop"
+    assert ("triton text" in capsys.readouterr().out) is not redact
 
 
+@pytest.mark.parametrize("redact", (False, True))
 def test_dispatch_text_completion_openai_with_usage(
     initialized_custom_stream_wrapper: CustomStreamWrapper,
+    redact: bool,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
 ):
     """text-completion-openai chunks expose choices[].text and an optional usage
     block that the dispatch lifts onto the model response."""
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    monkeypatch.setattr(litellm, "set_verbose", True)
     initialized_custom_stream_wrapper.custom_llm_provider = "text-completion-openai"
 
     class _Choice:
@@ -4026,6 +4053,7 @@ def test_dispatch_text_completion_openai_with_usage(
     assert initialized_custom_stream_wrapper.received_finish_reason == "stop"
     assert model_response.usage.prompt_tokens == 5
     assert model_response.usage.total_tokens == 8
+    assert ("oai text" in capsys.readouterr().out) is not redact
 
 
 @pytest.mark.parametrize("custom_llm_provider", ["text-completion-openai", "azure_text"])
@@ -8248,3 +8276,48 @@ def test_chunk_creator_usage_only_chunk_with_dict_usage_keeps_details(
     assert usage.prompt_tokens_details.cached_tokens == 80
     assert usage.completion_tokens_details is not None
     assert usage.completion_tokens_details.reasoning_tokens == 5
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+def test_legacy_stream_verbose_output_obeys_diagnostic_privacy(monkeypatch, capsys, enabled):
+    from litellm.litellm_core_utils.streaming_handler import print_verbose
+
+    monkeypatch.setattr(litellm, "set_verbose", True)
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", enabled)
+    print_verbose("chunk: %s", "PRIVATE_STREAM_CONTENT")
+    assert capsys.readouterr().out.strip() == ("redacted-by-litellm" if enabled else "chunk: PRIVATE_STREAM_CONTENT")
+
+
+@pytest.mark.parametrize("redact", (False, True))
+def test_predibase_stream_privacy_preserves_content_and_finish(
+    initialized_custom_stream_wrapper: CustomStreamWrapper,
+    redact: bool,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    monkeypatch.setattr(litellm, "set_verbose", True)
+    initialized_custom_stream_wrapper.custom_llm_provider = "predibase"
+    result, _, completion_obj = _run_dispatch(
+        initialized_custom_stream_wrapper,
+        b'data: {"token": {"text": "PRIVATE_PREDIBASE_REPLY"}, "details": {"finish_reason": "stop"}}',
+    )
+    assert isinstance(result, _ProviderChunkParsed)
+    assert completion_obj["content"] == "PRIVATE_PREDIBASE_REPLY"
+    assert initialized_custom_stream_wrapper.received_finish_reason == "stop"
+    assert ("PRIVATE_PREDIBASE_REPLY" in capsys.readouterr().out) is not redact
+
+
+@pytest.mark.parametrize("redact", (False, True))
+def test_triton_invalid_stream_chunk_keeps_error_out_of_private_diagnostics(
+    initialized_custom_stream_wrapper: CustomStreamWrapper,
+    redact: bool,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    monkeypatch.setattr(litellm, "set_verbose", True)
+    initialized_custom_stream_wrapper.custom_llm_provider = "triton"
+    with pytest.raises(ValueError, match=r"Unable to parse response.*PRIVATE_TRITON_CHUNK"):
+        _run_dispatch(initialized_custom_stream_wrapper, ["PRIVATE_TRITON_CHUNK"])
+    assert ("PRIVATE_TRITON_CHUNK" in capsys.readouterr().out) is not redact
diff --git a/tests/unit/llms/codestral/completion/test_codestral_completion_handler.py b/tests/unit/llms/codestral/completion/test_codestral_completion_handler.py
index 9ce50d8ceca..5d6b1cf5d27 100644
--- a/tests/unit/llms/codestral/completion/test_codestral_completion_handler.py
+++ b/tests/unit/llms/codestral/completion/test_codestral_completion_handler.py
@@ -48,3 +48,32 @@ async def test_fim_stream_keeps_unicode_line_separators_inside_a_delta(
 
     assert texts == [_LINE_SEPARATED_TEXT, None]
     assert json.loads(route.calls.last.request.content)["stream"] is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("privacy", [False, True])
+async def test_fim_response_content_is_preserved_when_diagnostics_are_private(
+    privacy: bool,
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", privacy)
+    monkeypatch.setattr(litellm, "set_verbose", True)
+    route: Final = respx_mock.post("https://codestral.mistral.ai/v1/fim/completions").mock(
+        return_value=httpx.Response(200, json={
+            "id": "fim-private", "object": "text_completion", "created": 1, "model": "codestral-latest",
+            "choices": [{"index": 0, "message": {"content": "PRIVATE_FIM_REPLY"}, "finish_reason": "stop"}],
+            "usage": {"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5},
+        })
+    )
+
+    response: Final = await litellm.atext_completion(
+        model="text-completion-codestral/codestral-latest",
+        prompt="PRIVATE_FIM_PROMPT", api_key="test-codestral-key", max_retries=0,
+    )
+
+    assert response.choices[0].text == "PRIVATE_FIM_REPLY"
+    assert json.loads(route.calls.last.request.content)["prompt"] == "PRIVATE_FIM_PROMPT"
+    assert ("PRIVATE_FIM_REPLY" in capsys.readouterr().out) is not privacy
diff --git a/tests/unit/proxy/common_utils/test_debug_utils.py b/tests/unit/proxy/common_utils/test_debug_utils.py
index 456be4e8c00..2a70bf5fe2a 100644
--- a/tests/unit/proxy/common_utils/test_debug_utils.py
+++ b/tests/unit/proxy/common_utils/test_debug_utils.py
@@ -198,3 +198,35 @@ def test_memory_routes_count_prompt_caching_pins(
 
     assert response.status_code == 200
     assert pin_count(response.json()) == expected
+
+
+@pytest.mark.asyncio
+async def test_otel_span_debug_privacy_preserves_api_result(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    from typing import Final
+
+    import litellm
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+    from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+    from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig
+    from litellm.proxy.common_utils.debug_utils import get_otel_spans
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    exporter: Final = InMemorySpanExporter()
+    provider: Final = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    logger: Final = OpenTelemetry(config=OpenTelemetryConfig(exporter=exporter), tracer_provider=provider)
+    monkeypatch.setattr(proxy_server, "open_telemetry_logger", logger)
+    try:
+        with provider.get_tracer("privacy-test").start_as_current_span("PRIVATE_SPAN_42896") as span:
+            span.set_attribute("gen_ai.input.messages", "PRIVATE_MESSAGE_42896")
+        result: Final = await get_otel_spans()
+        output: Final = capsys.readouterr().out
+        assert result["otel_spans"] == ["PRIVATE_SPAN_42896"]
+        assert "redacted-by-litellm" in output
+        assert "PRIVATE" not in output
+    finally:
+        provider.shutdown()
diff --git a/tests/unit/proxy/proxy_server/test_proxy_config.py b/tests/unit/proxy/proxy_server/test_proxy_config.py
index 49f9e99cf48..d74de84cfff 100644
--- a/tests/unit/proxy/proxy_server/test_proxy_config.py
+++ b/tests/unit/proxy/proxy_server/test_proxy_config.py
@@ -6271,3 +6271,160 @@ async def test_proxy_config_validates_advertised_mcp_versions_at_load(tmp_path,
         return
     with pytest.raises(ValidationError):
         await ProxyConfig().load_config(router=None, config_file_path=str(config))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("privacy", [False, True])
+async def test_proxy_config_load_applies_diagnostic_privacy(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, privacy: bool
+) -> None:
+    config_file: Final = tmp_path / "privacy.yaml"
+    config_file.write_text(
+        "model_list: []\nlitellm_settings:\n  redact_messages_in_debug_logs: " + str(privacy).lower() + "\n"
+    )
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", not privacy)
+    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+    monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", False)
+    monkeypatch.delenv("LITELLM_CONFIG_BUCKET_NAME", raising=False)
+
+    await ProxyConfig().load_config(router=None, config_file_path=str(config_file))
+
+    assert litellm.redact_messages_in_debug_logs is privacy
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("file_privacy", [None, False, True])
+@pytest.mark.parametrize("db_privacy", [False, True, "invalid-boolean"])
+@pytest.mark.parametrize("env_reference", [False, True])
+async def test_proxy_config_privacy_precedes_database_settings_diagnostics(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+    file_privacy: bool | None,
+    db_privacy: bool | str,
+    env_reference: bool,
+) -> None:
+    from litellm.caching.dual_cache import DualCache
+    from litellm.proxy import proxy_server
+    from litellm.proxy import utils as proxy_utils
+
+    monkeypatch.setenv("LITELLM_QA_DB_PRIVACY", str(db_privacy).lower())
+    monkeypatch.setenv("LITELLM_QA_FILE_PRIVACY", str(file_privacy).lower())
+    file_value: Final = "os.environ/LITELLM_QA_FILE_PRIVACY" if env_reference else file_privacy
+    remote_module: Final = "s3://example-bucket/PRIVATE_DB_CONTENT.py"
+    stored: Final = {
+        "general_settings": {
+            "alerting": ["PRIVATE_DB_CONTENT"],
+            "custom_auth": remote_module,
+            "litellm_jwtauth": {"custom_validate": remote_module},
+            "pass_through_endpoints": [{"path": "/PRIVATE_DB_CONTENT", "target": remote_module}],
+            "PRIVATE_DB_SETTING": "database-value",
+        },
+        "router_settings": {"model_group_alias": {"PRIVATE_DB_CONTENT": "qa-model"}},
+        "litellm_settings": {
+            "redact_messages_in_debug_logs": "os.environ/LITELLM_QA_DB_PRIVACY" if env_reference else db_privacy,
+            "model_alias_map": {"PRIVATE_DB_CONTENT": "qa-model"},
+            "callbacks": [remote_module, "prometheus"],
+            "custom_provider_map": [{"provider": "qa", "custom_handler": remote_module}],
+            "guardrails": [{"qa": {"callbacks": [remote_module], "guardrail": remote_module}}],
+        },
+    }
+
+    async def get_generic_data(key: str, value: str, table_name: str) -> SimpleNamespace | None:
+        assert (key, table_name) == ("param_name", "config")
+        row: Final = stored.get(value)
+        return SimpleNamespace(param_name=value, param_value=row) if row is not None else None
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", False)
+    monkeypatch.setattr(proxy_server, "prisma_client", SimpleNamespace(get_generic_data=get_generic_data))
+    monkeypatch.setattr(proxy_server, "store_model_in_db", True)
+    monkeypatch.setattr(proxy_utils, "litellm_config_cache", DualCache())
+    monkeypatch.delenv("LITELLM_CONFIG_BUCKET_NAME", raising=False)
+    config_file: Final = tmp_path / "db-privacy.yaml"
+    config_file.write_text(
+        json.dumps(
+            {
+                "litellm_settings": {} if file_privacy is None else {"redact_messages_in_debug_logs": file_value},
+                "general_settings": {"PRIVATE_DB_SETTING": "file-value"},
+            }
+        )
+    )
+
+    enabled: Final = db_privacy if file_privacy is None else file_privacy
+    if isinstance(enabled, str):
+        with caplog.at_level(logging.DEBUG, logger="LiteLLM Proxy"), pytest.raises(ValidationError):
+            await ProxyConfig().get_config(str(config_file))
+        assert "PRIVATE_DB_CONTENT" not in caplog.text
+        assert litellm.redact_messages_in_debug_logs is False
+        return
+
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM Proxy"):
+        loaded: Final = await ProxyConfig().get_config(str(config_file))
+
+    assert ("PRIVATE_DB_CONTENT" in caplog.text) is not enabled
+    assert litellm.redact_messages_in_debug_logs is enabled
+    assert loaded["litellm_settings"]["redact_messages_in_debug_logs"] is enabled
+    assert loaded["litellm_settings"]["model_alias_map"] == {"PRIVATE_DB_CONTENT": "qa-model"}
+    assert loaded["general_settings"] == {
+        "alerting": ["PRIVATE_DB_CONTENT"],
+        "litellm_jwtauth": {"custom_validate": None},
+        "pass_through_endpoints": [{"path": "/PRIVATE_DB_CONTENT", "target": None}],
+        "PRIVATE_DB_SETTING": "file-value",
+    }
+    assert loaded["litellm_settings"]["callbacks"] == ["prometheus"]
+    assert loaded["litellm_settings"]["custom_provider_map"] == [{"provider": "qa", "custom_handler": None}]
+    assert loaded["litellm_settings"]["guardrails"] == [{"qa": {"callbacks": [], "guardrail": None}}]
+    assert stored["general_settings"]["custom_auth"] == remote_module
+    assert any(record.levelno == logging.WARNING for record in caplog.records)
+    assert loaded["router_settings"] == stored["router_settings"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("privacy", [False, True])
+async def test_shadow_warning_survives_failed_config_load_and_emits_once(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, privacy: bool
+) -> None:
+    from litellm.caching.dual_cache import DualCache
+    from litellm.proxy import proxy_server
+    from litellm.proxy import utils as proxy_utils
+
+    async def get_generic_data(key: str, value: str, table_name: str) -> SimpleNamespace | None:
+        assert (key, table_name) == ("param_name", "config")
+        return (
+            SimpleNamespace(param_name=value, param_value={"max_parallel_requests": 2})
+            if value == "general_settings"
+            else None
+        )
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", False)
+    monkeypatch.setattr(proxy_server, "prisma_client", SimpleNamespace(get_generic_data=get_generic_data))
+    monkeypatch.setattr(proxy_server, "store_model_in_db", True)
+    monkeypatch.setattr(proxy_utils, "litellm_config_cache", DualCache())
+    monkeypatch.delenv("LITELLM_CONFIG_BUCKET_NAME", raising=False)
+    monkeypatch.setenv("LITELLM_QA_SHADOW_PRIVACY", "invalid-boolean")
+    config_file: Final = tmp_path / "shadow-warning.yaml"
+    config_file.write_text(
+        json.dumps(
+            {
+                "litellm_settings": {"redact_messages_in_debug_logs": "os.environ/LITELLM_QA_SHADOW_PRIVACY"},
+                "general_settings": {"max_parallel_requests": 5},
+            }
+        )
+    )
+    config: Final = ProxyConfig()
+    caplog.set_level(logging.WARNING, logger="LiteLLM Proxy")
+
+    with pytest.raises(ValidationError):
+        await config.get_config(str(config_file))
+    assert caplog.records == []
+
+    monkeypatch.setenv("LITELLM_QA_SHADOW_PRIVACY", str(privacy).lower())
+    loaded: Final = await config.get_config(str(config_file))
+    assert loaded["general_settings"]["max_parallel_requests"] == 5
+    assert len(caplog.records) == 1
+    assert caplog.records[0].levelno == logging.WARNING
+    assert ("max_parallel_requests" in caplog.records[0].getMessage()) is not privacy
+
+    caplog.clear()
+    await config.get_config(str(config_file))
+    assert caplog.records == []
diff --git a/tests/unit/proxy/test_common_request_processing.py b/tests/unit/proxy/test_common_request_processing.py
index aeab7a3d348..2d8349367c6 100644
--- a/tests/unit/proxy/test_common_request_processing.py
+++ b/tests/unit/proxy/test_common_request_processing.py
@@ -12519,6 +12519,218 @@ def test_maybe_get_model_id_returns_none_before_the_router_picked_a_deployment()
     assert ProxyBaseLLMRequestProcessing(data={"metadata": {}}).maybe_get_model_id(None) is None
 
 
+def test_debug_message_privacy_does_not_serialize_proxy_payload(monkeypatch: pytest.MonkeyPatch) -> None:
+    class Unrenderable:
+        def __str__(self) -> str:
+            raise AssertionError("diagnostics must not serialize request content")
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    content: Final = Unrenderable()
+    request: Final = ProxyBaseLLMRequestProcessing(data={"messages": [{"role": "user", "content": content}]})
+    request._debug_log_request_payload()
+    assert request.data["messages"][0]["content"] is content
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("privacy", [False, True])
+async def test_debug_message_privacy_preserves_stream_content(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, privacy: bool
+) -> None:
+    import logging
+
+    from litellm._logging import verbose_proxy_logger
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", privacy)
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(litellm, "success_callback", [])
+    monkeypatch.setattr(litellm, "_async_success_callback", [])
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False)
+    monkeypatch.setattr(verbose_proxy_logger, "level", logging.DEBUG)
+    ProxyLogging._callback_capabilities_cache.clear()
+    chunks: Final = ("data: PRIVATE_STREAM_INPUT_42896\n\n", "data: [DONE]\n\n")
+
+    async def upstream() -> AsyncGenerator[str, None]:
+        for chunk in chunks:
+            yield chunk
+
+    def serialize(chunk: object) -> str:
+        assert isinstance(chunk, str)
+        return chunk
+
+    stream: Final = ProxyBaseLLMRequestProcessing.async_streaming_data_generator(
+        response=upstream(),
+        user_api_key_dict=ProxyUserAPIKeyAuth(),
+        request_data={"model": "example-model"},
+        proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache()),
+        serialize_chunk=serialize,
+        serialize_error=lambda error: str(error),
+    )
+    observed: Final = tuple([chunk async for chunk in stream])
+    assert observed == chunks
+    messages: Final = tuple(
+        record.getMessage() for record in caplog.records if record.name == verbose_proxy_logger.name
+    )
+    if privacy:
+        assert any("event=stream_completed" in message for message in messages)
+        assert all("PRIVATE_STREAM_INPUT_42896" not in message for message in messages)
+    else:
+        assert any("PRIVATE_STREAM_INPUT_42896" in message for message in messages)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("privacy", [False, True])
+@pytest.mark.parametrize("http_error", [False, True])
+async def test_stream_failure_retains_error_status_without_debug_logging(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, privacy: bool, http_error: bool
+) -> None:
+    import logging
+
+    from litellm._logging import verbose_proxy_logger
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", privacy)
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(litellm, "success_callback", [])
+    monkeypatch.setattr(litellm, "_async_success_callback", [])
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False)
+    monkeypatch.setattr(verbose_proxy_logger, "level", logging.ERROR)
+    ProxyLogging._callback_capabilities_cache.clear()
+    failure: Final = (
+        HTTPException(status_code=429, detail="PRIVATE_STREAM_FAILURE_42896")
+        if http_error
+        else RuntimeError("PRIVATE_STREAM_FAILURE_42896")
+    )
+
+    async def upstream() -> AsyncGenerator[str, None]:
+        yield "data: started\n\n"
+        raise failure
+
+    stream: Final = ProxyBaseLLMRequestProcessing.async_streaming_data_generator(
+        response=upstream(),
+        user_api_key_dict=ProxyUserAPIKeyAuth(),
+        request_data={"model": "example-model"},
+        proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache()),
+        serialize_chunk=str,
+        serialize_error=str,
+    )
+    assert await anext(stream) == "data: started\n\n"
+    if http_error:
+        with pytest.raises(HTTPException) as raised:
+            await anext(stream)
+        assert raised.value is failure
+    else:
+        remaining: Final = tuple([chunk async for chunk in stream])
+        assert len(remaining) == 1
+        assert "PRIVATE_STREAM_FAILURE_42896" in remaining[0]
+    messages: Final = tuple(record.getMessage() for record in caplog.records if record.name == verbose_proxy_logger.name)
+    if privacy:
+        assert any(f"event=request_failed status={429 if http_error else 500}" in message for message in messages)
+        assert all("PRIVATE_STREAM_FAILURE_42896" not in message for message in messages)
+    else:
+        assert any("PRIVATE_STREAM_FAILURE_42896" in message for message in messages)
+        assert all("event=request_failed" not in message for message in messages)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("privacy", [False, True])
+async def test_nonstream_response_keeps_content_with_private_ready_diagnostic(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, privacy: bool
+) -> None:
+    import logging
+
+    import respx
+
+    from litellm._logging import verbose_proxy_logger
+    from litellm.proxy.proxy_server import ProxyConfig as ServerProxyConfig
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", privacy)
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "success_callback", [])
+    monkeypatch.setattr(litellm, "_async_success_callback", [])
+    monkeypatch.setattr(verbose_proxy_logger, "level", logging.DEBUG)
+    ProxyLogging._callback_capabilities_cache.clear()
+    content: Final = "PRIVATE_NONSTREAM_RESPONSE_42896"
+    processor: Final = ProxyBaseLLMRequestProcessing(
+        data={
+            "model": "openai/gpt-4o-mini",
+            "api_base": "https://example.com/v1",
+            "api_key": "test-key",
+            "messages": [{"role": "user", "content": "PRIVATE_NONSTREAM_INPUT_42896"}],
+        }
+    )
+    with respx.mock() as upstream:
+        upstream.post("https://example.com/v1/chat/completions").respond(
+            200,
+            json={
+                "id": "chatcmpl-example",
+                "object": "chat.completion",
+                "created": 1700000000,
+                "model": "gpt-4o-mini",
+                "choices": [
+                    {"index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop"}
+                ],
+                "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+            },
+        )
+        response: Final = await processor.base_process_llm_request(
+            request=Request(
+                {"type": "http", "method": "POST", "path": "/v1/chat/completions", "headers": [], "query_string": b""}
+            ),
+            fastapi_response=Response(),
+            user_api_key_dict=ProxyUserAPIKeyAuth(),
+            route_type="acompletion",
+            proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache()),
+            general_settings={},
+            proxy_config=ServerProxyConfig(),
+        )
+    assert isinstance(response, litellm.ModelResponse)
+    assert isinstance(response.choices[0], litellm.Choices)
+    assert response.choices[0].message.content == content
+    assert response.choices[0].finish_reason == "stop"
+    messages: Final = tuple(
+        record.getMessage() for record in caplog.records if record.name == verbose_proxy_logger.name
+    )
+    if privacy:
+        assert any("event=response_ready status=200 elapsed_seconds=" in message for message in messages)
+        assert all("PRIVATE_NONSTREAM_" not in message for message in messages)
+    else:
+        assert all("event=response_ready" not in message for message in messages)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("privacy", [False, True])
+async def test_nonstream_failure_preserves_client_error_and_safe_status_without_debug(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, privacy: bool
+) -> None:
+    import logging
+
+    from litellm._logging import verbose_proxy_logger
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", privacy)
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(verbose_proxy_logger, "level", logging.ERROR)
+    ProxyLogging._callback_capabilities_cache.clear()
+    processor: Final = ProxyBaseLLMRequestProcessing(data={"model": "example-model"})
+    failure: Final = HTTPException(status_code=429, detail="PRIVATE_NONSTREAM_FAILURE_42896")
+    with pytest.raises(ProxyException) as raised:
+        await processor.handle_llm_api_exception(
+            e=failure,
+            user_api_key_dict=ProxyUserAPIKeyAuth(),
+            proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache()),
+        )
+    assert raised.value.code == "429"
+    assert raised.value.message == "PRIVATE_NONSTREAM_FAILURE_42896"
+    messages: Final = tuple(
+        record.getMessage() for record in caplog.records if record.name == verbose_proxy_logger.name
+    )
+    if privacy:
+        assert any("event=request_failed status=429" in message for message in messages)
+        assert all("PRIVATE_NONSTREAM_FAILURE_42896" not in message for message in messages)
+    else:
+        assert any("PRIVATE_NONSTREAM_FAILURE_42896" in message for message in messages)
+        assert all("event=request_failed" not in message for message in messages)
+
+
 _DOWNSTREAM_GUARDRAIL_FIELDS: Final = MappingProxyType(
     {"guardrail_name": "block-secret-project", "guardrail_mode": "pre_call", "pattern": "secret_project_codename"}
 )
diff --git a/tests/unit/proxy/test_post_call_rules.py b/tests/unit/proxy/test_post_call_rules.py
new file mode 100644
index 00000000000..55ee5142bcf
--- /dev/null
+++ b/tests/unit/proxy/test_post_call_rules.py
@@ -0,0 +1,26 @@
+from typing import Final
+
+import pytest
+
+import litellm
+from litellm.proxy.post_call_rules import post_response_rule
+
+
+@pytest.mark.parametrize("redact", (False, True))
+@pytest.mark.parametrize("long_response", (False, True))
+def test_diagnostic_privacy_preserves_response_rule_decision(
+    redact: bool,
+    long_response: bool,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    marker: Final = "PRIVATE_RULE_INPUT"
+    response: Final = marker * (20 if long_response else 1)
+    expected: Final = (
+        {"decision": True}
+        if long_response
+        else {"decision": False, "message": "This violates LiteLLM Proxy Rules. Response too short"}
+    )
+    assert post_response_rule(response) == expected
+    assert (marker in capsys.readouterr().out) is not redact
diff --git a/tests/unit/proxy/test_proxy_server.py b/tests/unit/proxy/test_proxy_server.py
index c835e1e27ea..55252239bfe 100644
--- a/tests/unit/proxy/test_proxy_server.py
+++ b/tests/unit/proxy/test_proxy_server.py
@@ -4072,6 +4072,70 @@ async def test_model_list_return_wildcard_routes_defaults_to_general_setting(
     assert ("openai/*" in listed) is wildcard_listed, listed
 
 
+@pytest.mark.parametrize("enabled", [True, False])
+async def test_yaml_config_applies_diagnostic_privacy_before_model_initialization(
+    tmp_path, monkeypatch: pytest.MonkeyPatch, enabled: bool
+) -> None:
+    from litellm.proxy.proxy_server import ProxyConfig
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", not enabled)
+    config_file: Final = tmp_path / "config.yaml"
+    config_file.write_text("litellm_settings:\n  redact_messages_in_debug_logs: " + str(enabled).lower() + "\n")
+    config: Final = await ProxyConfig().get_config(str(config_file))
+    assert litellm.redact_messages_in_debug_logs is enabled
+    assert config["litellm_settings"]["redact_messages_in_debug_logs"] is enabled
+
+
+@pytest.mark.parametrize("as_json", [False, True])
+@pytest.mark.parametrize("enabled", [False, True])
+def test_worker_config_privacy_applies_before_global_setting(
+    monkeypatch: pytest.MonkeyPatch, as_json: bool, enabled: bool
+) -> None:
+    from litellm.proxy.proxy_server import _redact_worker_config_for_logging
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", False)
+    config: Final = {
+        "litellm_settings": {"redact_messages_in_debug_logs": enabled},
+        "model_list": [{"litellm_params": {"messages": [{"content": "PRIVATE_SYSTEM_PROMPT"}]}}],
+    }
+    result: Final = _redact_worker_config_for_logging(json.dumps(config) if as_json else config)
+    if enabled:
+        assert result == "redacted-by-litellm"
+    else:
+        assert "PRIVATE_SYSTEM_PROMPT" in str(result)
+
+
+@pytest.mark.parametrize("as_json", [False, True])
+def test_worker_config_invalid_privacy_setting_does_not_expose_content(
+    monkeypatch: pytest.MonkeyPatch, as_json: bool
+) -> None:
+    from litellm.proxy.proxy_server import _redact_worker_config_for_logging
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", False)
+    config: Final = {
+        "litellm_settings": {"redact_messages_in_debug_logs": "invalid-boolean"},
+        "model_list": [{"litellm_params": {"messages": [{"content": "PRIVATE_SYSTEM_PROMPT"}]}}],
+    }
+    original: Final = json.dumps(config)
+    result: Final = _redact_worker_config_for_logging(original if as_json else config)
+
+    assert result == "redacted-by-litellm"
+    assert json.dumps(config) == original
+    assert litellm.redact_messages_in_debug_logs is False
+
+
+@pytest.mark.parametrize("worker_config", ['{"messages": "PRIVATE_SYSTEM_PROMPT"', None])
+@pytest.mark.parametrize("enabled", [False, True])
+def test_worker_config_global_privacy_covers_unparseable_config(
+    monkeypatch: pytest.MonkeyPatch, worker_config: str | None, enabled: bool
+) -> None:
+    from litellm.proxy.proxy_server import _redact_worker_config_for_logging
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", enabled)
+
+    assert _redact_worker_config_for_logging(worker_config) == ("redacted-by-litellm" if enabled else worker_config)
+
+
 @pytest.mark.asyncio
 async def test_queue_request_labels_an_upstream_failure_a_server_error_not_an_auth_error(
     monkeypatch: pytest.MonkeyPatch,
diff --git a/tests/unit/responses/mcp/test_mcp_streaming_iterator.py b/tests/unit/responses/mcp/test_mcp_streaming_iterator.py
index f745b5bfadf..74026c0d0db 100644
--- a/tests/unit/responses/mcp/test_mcp_streaming_iterator.py
+++ b/tests/unit/responses/mcp/test_mcp_streaming_iterator.py
@@ -8,7 +8,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 from mcp.types import CallToolResult, TextContent
 
-import litellm  # noqa: F401 - ensures litellm.responses.main is registered in sys.modules
+import litellm
 from litellm.responses.mcp.mcp_streaming_iterator import (
     MCPEnhancedStreamingIterator,
 )
@@ -293,14 +293,18 @@ async def test_continuation_id_is_final_round_not_interim_tool_call(monkeypatch)
     assert completed[-1].response.id != "resp-interim"
 
 
+@pytest.mark.parametrize("redact", (False, True))
 @pytest.mark.asyncio
-async def test_follow_up_call_failure_emits_terminal_error_event(monkeypatch):
+async def test_follow_up_call_failure_emits_terminal_error_event(
+    monkeypatch: pytest.MonkeyPatch, redact: bool, capsys: pytest.CaptureFixture[str]
+) -> None:
     """
     Regression test for the silent-swallow path: when the follow-up LLM call
     raises, the stream must emit a terminal `error` event instead of ending
     silently after the tool events (which surfaced to clients as a successful
     but empty completion).
     """
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
     _mock_mcp_environment(monkeypatch)
 
     aresponses_mock = AsyncMock(side_effect=RuntimeError("boom from provider"))
@@ -319,16 +323,21 @@ async def test_follow_up_call_failure_emits_terminal_error_event(monkeypatch):
     assert len(error_events) == 1
     assert error_events[0].error.type == "mcp_gateway_error"
     assert "boom from provider" in error_events[0].error.message
+    assert ("boom from provider" in capsys.readouterr().err) is not redact
 
 
+@pytest.mark.parametrize("redact", (False, True))
 @pytest.mark.asyncio
-async def test_initial_call_failure_is_stashed_for_eager_reraise(monkeypatch):
+async def test_initial_call_failure_is_stashed_for_eager_reraise(
+    monkeypatch: pytest.MonkeyPatch, redact: bool, capsys: pytest.CaptureFixture[str]
+) -> None:
     """
     Regression test: a failing initial LLM call must be stashed as
     `_initial_creation_error` so `aresponses_api_with_mcp` can re-raise it as a
     real 4xx before any SSE bytes are written, instead of returning HTTP 200
     with an empty stream.
     """
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
     _mock_mcp_environment(monkeypatch)
 
     aresponses_mock = AsyncMock(side_effect=RuntimeError("initial boom"))
@@ -339,6 +348,7 @@ async def test_initial_call_failure_is_stashed_for_eager_reraise(monkeypatch):
 
     assert iterator._initial_creation_error is not None
     assert "initial boom" in str(iterator._initial_creation_error)
+    assert ("initial boom" in capsys.readouterr().err) is not redact
 
 
 def _reasoning_item(encrypted_content: str):
@@ -581,3 +591,40 @@ async def test_stream_without_auto_execute_is_forwarded_unchanged(monkeypatch):
     assert [c.output_index for c in chunks if hasattr(c, "output_index")] == [0, 0]
     assert [c.sequence_number for c in chunks if hasattr(c, "sequence_number")] == [0, 1, 2]
     aresponses_mock.assert_not_called()
+
+
+@pytest.mark.parametrize("redact", (False, True))
+@pytest.mark.asyncio
+async def test_malformed_provider_tool_arguments_keep_error_diagnostics_private(
+    monkeypatch: pytest.MonkeyPatch, redact: bool, capsys: pytest.CaptureFixture[str]
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    response: Final = ResponsesAPIResponse(
+        id="resp-malformed-arguments",
+        created_at=0,
+        output=[
+            {
+                "type": "function_call",
+                "call_id": "call-1",
+                "name": "read_wiki_contents",
+                "arguments": {"text": "PRIVATE_TOOL_ARGUMENT"},
+            }
+        ],
+    )
+    iterator: Final = _make_iterator([])
+    iterator.collected_response = response
+
+    await iterator._generate_tool_execution_events()
+
+    assert iterator.tool_results == []
+    assert iterator._tool_results_for_response is response
+    assert iterator.collected_response.output == [
+        {
+            "type": "function_call",
+            "call_id": "call-1",
+            "name": "read_wiki_contents",
+            "arguments": {"text": "PRIVATE_TOOL_ARGUMENT"},
+        }
+    ]
+    assert [event.type for event in iterator.tool_execution_events] == [ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED]
+    assert ("PRIVATE_TOOL_ARGUMENT" in capsys.readouterr().err) is not redact
diff --git a/tests/unit/rust_bridge/test_logger.py b/tests/unit/rust_bridge/test_logger.py
index 4745aca9d50..580edec6fc5 100644
--- a/tests/unit/rust_bridge/test_logger.py
+++ b/tests/unit/rust_bridge/test_logger.py
@@ -132,3 +132,14 @@ def test_missing_native_diagnostic_processor_falls_back_before_record_mutation(
 def test_unsupported_unicode_uses_safe_python_redaction(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("LITELLM_RUST", "1")
     assert redact_secrets("broken\ud800 api_key=secret123") == "broken\ud800 REDACTED"
+
+
+@pytest.mark.parametrize("level", (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR))
+def test_native_diagnostics_do_not_format_or_emit_content_when_privacy_enabled(
+    level: int, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        assert not logger.enabled(level)
+        logger.emit(level, "private response", "native.rs", 1, "litellm_http", {"body": "private prompt"}, ("", ""))
+    assert caplog.records == []
diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py
index fcd835e93dd..a96b9abe784 100644
--- a/tests/unit/test_logging.py
+++ b/tests/unit/test_logging.py
@@ -1841,6 +1841,36 @@ def test_print_verbose_formats_args_only_when_set_verbose_is_on(monkeypatch, cap
     assert capsys.readouterr().out == "key k1 value cached-vector\n"
 
 
+@pytest.mark.parametrize("privacy", [False, True])
+@pytest.mark.parametrize("json_logs", [False, True])
+@pytest.mark.parametrize("warning", [False, True])
+@pytest.mark.parametrize("suppressed", [False, True])
+def test_startup_message_respects_diagnostic_privacy(
+    monkeypatch: pytest.MonkeyPatch,
+    capfd: pytest.CaptureFixture[str],
+    privacy: bool,
+    json_logs: bool,
+    warning: bool,
+    suppressed: bool,
+) -> None:
+    from litellm._logging import print_startup_warning
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", privacy)
+    monkeypatch.setattr(litellm, "json_logs", json_logs)
+    monkeypatch.setattr(litellm, "suppress_startup_output", suppressed)
+    emit: Final = print_startup_warning if warning else print_startup_message
+    emit("PRIVATE_CONFIGURED_MODEL_42896")
+    out, err = capfd.readouterr()
+    assert err == ""
+    if suppressed and not warning:
+        assert out == ""
+        return
+    message: Final = json.loads(out)["message"] if json_logs else out.strip()
+    assert message == ("redacted-by-litellm" if privacy else "PRIVATE_CONFIGURED_MODEL_42896")
+    if json_logs:
+        assert json.loads(out)["level"] == ("WARNING" if warning else "INFO")
+
+
 class _CapturedStream(StringIO):
     def __init__(self, tty: bool) -> None:
         super().__init__()
@@ -2118,3 +2148,168 @@ def test_print_startup_message_blank_spacer_emits_no_json_record(
     print_startup_message("\n\033[0m\n")
 
     assert capfd.readouterr() == ("", "")
+
+
+class _UnrenderableDiagnosticPayload:
+    def __str__(self) -> str:
+        raise AssertionError("diagnostic payload was rendered")
+
+    def __repr__(self) -> str:
+        raise AssertionError("diagnostic payload was rendered")
+
+
+@pytest.mark.parametrize("level", [logging.DEBUG, logging.WARNING, logging.ERROR])
+@pytest.mark.parametrize("credential_redaction", [False, True])
+def test_message_privacy_removes_payload_before_rendering(
+    monkeypatch: pytest.MonkeyPatch, level: int, credential_redaction: bool
+) -> None:
+    import litellm._logging as logging_module
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    monkeypatch.setattr(logging_module, "_ENABLE_SECRET_REDACTION", credential_redaction)
+    payload: Final = _UnrenderableDiagnosticPayload()
+    record: Final = logging.LogRecord("LiteLLM", level, __file__, 1, "payload=%s", (payload,), None)
+    record.exc_info = (ValueError, ValueError("PRIVATE_EXCEPTION"), None)
+    record.exc_text = "PRIVATE_CACHED_TRACEBACK"
+    record.stack_info = "PRIVATE_STACK"
+    record.payload = payload
+    record.color_message = "PRIVATE_COLOR"
+    record.message = "PRIVATE_CACHED_MESSAGE"
+    assert DiagnosticProcessingFilter().filter(record)
+    rendered: Final = JsonFormatter().format(record)
+    assert json.loads(rendered)["message"] == "redacted-by-litellm"
+    assert "PRIVATE" not in rendered
+    assert record.exc_info is None
+    assert record.args == ()
+    assert not hasattr(record, "payload")
+
+
+@pytest.mark.parametrize("spoof", [True, "safe", "redacted-by-litellm"])
+def test_message_privacy_cannot_be_disabled_by_log_extras(monkeypatch: pytest.MonkeyPatch, spoof: object) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    record: Final = logging.LogRecord("LiteLLM", logging.DEBUG, __file__, 1, "PRIVATE_PROMPT", (), None)
+    record.litellm_safe_diagnostic = spoof
+    assert DiagnosticProcessingFilter().filter(record)
+    assert record.getMessage() == "redacted-by-litellm"
+
+
+def test_message_privacy_disabled_preserves_record(monkeypatch: pytest.MonkeyPatch) -> None:
+    from litellm._logging import MessageContentFilter
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", False)
+    payload: Final = {"messages": [{"content": "PRIVATE_PROMPT"}]}
+    record: Final = logging.LogRecord("LiteLLM", logging.DEBUG, __file__, 1, "payload=%s", (payload,), None)
+    before: Final = dict(record.__dict__)
+    assert MessageContentFilter().filter(record)
+    assert record.__dict__ == before
+    assert record.args is before["args"]
+    assert "PRIVATE_PROMPT" in record.getMessage()
+
+
+def test_message_privacy_keeps_safe_operational_events(monkeypatch: pytest.MonkeyPatch) -> None:
+    from litellm._logging import MessageContentFilter, log_safe_diagnostic
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    stream: Final = StringIO()
+    handler: Final = logging.StreamHandler(stream)
+    handler.setFormatter(JsonFormatter())
+    handler.addFilter(MessageContentFilter())
+    logger: Final = logging.Logger("privacy-test", level=logging.DEBUG)
+    logger.addFilter(DiagnosticProcessingFilter())
+    logger.addHandler(handler)
+    log_safe_diagnostic(logger, "retry", status_code=429, elapsed_seconds=0.25, attempt=2)
+    result: Final = json.loads(stream.getvalue())
+    assert result["message"] == "event=retry status=429 elapsed_seconds=0.25 attempt=2 content=redacted-by-litellm"
+    assert "litellm_safe_diagnostic" not in result
+
+
+def test_message_privacy_suppresses_caller_correlation_fields(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    monkeypatch.setattr(litellm, "request_correlation_in_logs", True)
+    trace_token: Final = set_trace_id("PRIVATE_TRACE")
+    session_token: Final = set_session_id("PRIVATE_SESSION")
+    try:
+        record: Final = logging.LogRecord("LiteLLM", logging.DEBUG, __file__, 1, "PRIVATE_PROMPT", (), None)
+        assert DiagnosticProcessingFilter().filter(record)
+        assert CorrelationContextFilter().filter(record)
+        assert "PRIVATE" not in JsonFormatter().format(record)
+    finally:
+        trace_id_var.reset(trace_token)
+        session_id_var.reset(session_token)
+
+
+def test_json_formatter_alone_obeys_message_privacy(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    record: Final = logging.LogRecord(
+        "LiteLLM.child", logging.ERROR, __file__, 1, _UnrenderableDiagnosticPayload(), (), None
+    )
+    assert json.loads(JsonFormatter().format(record))["message"] == "redacted-by-litellm"
+
+
+def test_message_privacy_precedes_custom_handler(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    stream: Final = StringIO()
+    handler: Final = logging.StreamHandler(stream)
+    verbose_logger.addHandler(handler)
+    try:
+        verbose_logger.warning("PRIVATE_PROMPT %s", _UnrenderableDiagnosticPayload())
+        assert stream.getvalue().strip() == "redacted-by-litellm"
+    finally:
+        verbose_logger.removeHandler(handler)
+
+
+def test_message_privacy_does_not_change_application_records(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    record: Final = logging.LogRecord("application", logging.INFO, __file__, 1, "application says %s", ("hello",), None)
+    assert SecretRedactionFilter().filter(record)
+    assert json.loads(JsonFormatter().format(record))["message"] == "application says hello"
+
+
+def _raise_private_diagnostic_exception() -> None:
+    raise ValueError("PRIVATE_ASYNC_EXCEPTION")
+
+
+@pytest.mark.parametrize("module", ["litellm.background", "application.background"])
+def test_asyncio_message_privacy_is_scoped_to_litellm_tracebacks(monkeypatch: pytest.MonkeyPatch, module: str) -> None:
+    from types import FunctionType
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    fail: Final = FunctionType(_raise_private_diagnostic_exception.__code__, {"__name__": module})
+    try:
+        fail()
+    except ValueError:
+        record: Final = logging.LogRecord("asyncio", logging.ERROR, __file__, 1, "PRIVATE_TASK", (), sys.exc_info())
+        assert SecretRedactionFilter().filter(record)
+        rendered: Final = JsonFormatter().format(record)
+        if module.startswith("litellm."):
+            assert "PRIVATE" not in rendered
+        else:
+            assert "PRIVATE_TASK" in rendered
+            assert "PRIVATE_ASYNC_EXCEPTION" in rendered
+
+
+def test_private_diagnostics_support_custom_formatter_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
+    from litellm._logging import MessageContentFilter
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    record: Final = logging.LogRecord("LiteLLM", logging.ERROR, __file__, 1, "PRIVATE_PROMPT", (), None)
+    record.customer_metadata = "PRIVATE_METADATA"
+    assert MessageContentFilter().filter(record)
+    formatter: Final = logging.Formatter("%(levelname)s %(customer_metadata)s %(message)s", defaults={"customer_metadata": "-"})
+    assert formatter.format(record) == "ERROR - redacted-by-litellm"
+
+
+def test_private_access_logs_keep_formatter_contract_without_request_content(monkeypatch: pytest.MonkeyPatch) -> None:
+    from uvicorn.logging import AccessFormatter
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    record: Final = logging.LogRecord(
+        "uvicorn.access", logging.INFO, __file__, 1, "PRIVATE_TEMPLATE",
+        ("PRIVATE_CLIENT", "GET", "/v1/chat/completions?prompt=PRIVATE_PROMPT", "1.1", 200), None,
+    )
+    record.metadata = "PRIVATE_METADATA"
+    assert AccessLogRedactionFilter().filter(record)
+    formatted: Final = AccessFormatter('%(client_addr)s - "%(request_line)s" %(status_code)s', use_colors=False).format(record)
+    assert "PRIVATE" not in formatted
+    assert "200" in formatted
+    assert "PRIVATE" not in JsonFormatter().format(record)
diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py
index 571f890a4e6..1b0500e21e5 100644
--- a/tests/unit/test_main.py
+++ b/tests/unit/test_main.py
@@ -10809,3 +10809,39 @@ async def test_bedrock_strict_tools_limit_rejection_is_not_sent_again_when_no_to
 
     assert error is not None and "Too many strict tools" in str(error)
     assert len(sent) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("privacy", [False, True])
+async def test_async_streaming_keeps_chunks_without_rendering_private_content(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], privacy: bool
+) -> None:
+    from collections.abc import AsyncIterator
+
+    class Chunk:
+        content: Final = "PRIVATE_ASYNC_STREAM_42896"
+
+        def __str__(self) -> str:
+            if privacy:
+                raise AssertionError("private chunks must not be rendered for diagnostics")
+            return self.content
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", privacy)
+    monkeypatch.setattr(litellm, "set_verbose", True)
+    chunk: Final = Chunk()
+
+    async def upstream() -> AsyncIterator[Chunk]:
+        yield chunk
+
+    stream: Final = litellm_main._async_streaming(upstream(), "example-model", "openai", {})
+    observed: Final = tuple([item async for item in stream])
+
+    assert len(observed) == 1
+    assert observed[0] is chunk
+    assert chunk.content == "PRIVATE_ASYNC_STREAM_42896"
+    output: Final = capsys.readouterr().out
+    assert ("PRIVATE_ASYNC_STREAM_42896" in output) is not privacy
+    if privacy:
+        assert output.strip() == "redacted-by-litellm"
+    else:
+        assert "line in async streaming: PRIVATE_ASYNC_STREAM_42896" in output
diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py
index 243a20a2551..110ab810b8f 100644
--- a/tests/unit/test_utils.py
+++ b/tests/unit/test_utils.py
@@ -11294,6 +11294,39 @@ def test_get_optional_params_forwards_a_tool_config_cache_point_to_a_downstream_
     assert optional_params["extra_body"] == {"kept": True, "cache_control_injection_points": [point]}
 
 
+@pytest.mark.parametrize("redact", (False, True))
+@pytest.mark.parametrize("module_name", ("litellm.utils", "litellm.main", "litellm.caching.caching", "litellm.proxy.utils"))
+def test_legacy_verbose_stdout_obeys_diagnostic_privacy(
+    redact: bool,
+    module_name: str,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", redact)
+    monkeypatch.setattr(litellm, "set_verbose", True)
+    module: Final = importlib.import_module(module_name)
+    module.print_verbose("synthetic-private-prompt-42896")
+    stdout: Final = capsys.readouterr().out
+    assert ("synthetic-private-prompt-42896" in stdout) is not redact
+    assert ("redacted-by-litellm" in stdout) is redact
+
+
+def test_private_argument_diagnostic_does_not_render_request_objects(
+    capsys: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from litellm.utils import print_args_passed_to_litellm
+
+    class UnrenderableRequest:
+        def __repr__(self) -> str:
+            raise AssertionError("diagnostic attempted to render a private request")
+
+    monkeypatch.setattr(litellm, "redact_messages_in_debug_logs", True)
+    monkeypatch.setattr(litellm, "set_verbose", True)
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        print_args_passed_to_litellm(litellm.completion, (UnrenderableRequest(),), {"messages": UnrenderableRequest()})
+    assert capsys.readouterr().out.endswith("redacted-by-litellm\n")
+
+
 
 @pytest.mark.parametrize(
     ("provider", "expected_warnings"),