fix(logging): redact system_prompt when message logging is off

Include system_prompt in _redact_standard_logging_object and
CustomLogger.redact_standard_logging_payload_from_model_call_details
so list-form system blocks do not bypass turn_off_message_logging.

Also remove new docstring/comments flagged in PR review.
This commit is contained in:
Souravrajvi0 2026-08-10 17:28:54 +00:00
parent 850eada677
commit bc81ccf9d9
6 changed files with 60 additions and 15 deletions

View file

@ -885,6 +885,12 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
if "messages" not in (excluded_fields or []) and standard_logging_object_copy.get("messages") is not None:
standard_logging_object_copy["messages"] = [Message(content=redacted_str).model_dump()]
if (
"system_prompt" not in (excluded_fields or [])
and standard_logging_object_copy.get("system_prompt") is not None
):
standard_logging_object_copy["system_prompt"] = redacted_str
if "response" not in (excluded_fields or []) and standard_logging_object_copy.get("response") is not None:
response = standard_logging_object_copy["response"]
# Check if this is a ResponsesAPIResponse (has "output" field)

View file

@ -4557,13 +4557,6 @@ class StandardLoggingPayloadSetup:
@staticmethod
def get_system_prompt_from_kwargs(kwargs: Optional[Dict] = None) -> Optional[Union[str, list, dict]]:
"""
Return the system prompt kwargs as sent by the client, without reshaping.
Coalesces the kwarg names used across call paths (Vertex Gemini, Responses API,
Anthropic Messages). Uses `is not None` checks so falsy values like [] do not
fall through to a different kwarg.
"""
if kwargs is None:
return None

View file

@ -169,6 +169,9 @@ def _redact_standard_logging_object(model_call_details: dict):
if standard_logging_object.get("messages") is not None:
standard_logging_object["messages"] = [{"role": "user", "content": redacted_str}]
if standard_logging_object.get("system_prompt") is not None:
standard_logging_object["system_prompt"] = redacted_str
response = standard_logging_object.get("response")
if response is not None:
if isinstance(response, dict) and "output" in response:

View file

@ -62,6 +62,7 @@ def create_sample_standard_logging_payload() -> Dict:
"requester_ip_address": None,
"user_agent": None,
"messages": [{"role": "user", "content": "Hello, this is sensitive data!"}],
"system_prompt": [{"type": "text", "text": "sensitive system prompt"}],
"response": {
"choices": [{"message": {"content": "This is a sensitive response!"}}]
},
@ -226,6 +227,7 @@ class TestStandardLoggingPayloadExcludedFields:
assert (
result["standard_logging_object"]["messages"][0]["content"] == redacted_str
)
assert result["standard_logging_object"]["system_prompt"] == redacted_str
assert (
result["standard_logging_object"]["response"]["choices"][0]["message"][
"content"

View file

@ -2166,7 +2166,6 @@ def test_append_system_prompt_messages():
def test_get_system_prompt_from_kwargs():
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
# Anthropic Messages list-form system blocks
system_blocks = [
{"type": "text", "text": "SHAPE-SECRET", "cache_control": {"type": "ephemeral"}},
]
@ -2175,22 +2174,18 @@ def test_get_system_prompt_from_kwargs():
)
assert result == system_blocks
# String system kwarg
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs={"system": "Be helpful"})
assert result == "Be helpful"
# Responses API instructions
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs={"instructions": "Follow policy"})
assert result == "Follow policy"
# Vertex Gemini system_instructions
gemini_system = [{"role": "system", "content": "Be concise."}]
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
kwargs={"system_instructions": gemini_system}
)
assert result == gemini_system
# system_instructions wins over instructions and system
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
kwargs={
"system_instructions": "From Gemini",
@ -2200,19 +2195,16 @@ def test_get_system_prompt_from_kwargs():
)
assert result == "From Gemini"
# Empty list should not fall through to instructions
result = StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(
kwargs={"system_instructions": [], "instructions": "From Responses"}
)
assert result == []
# No system kwargs
assert StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs={}) is None
assert StandardLoggingPayloadSetup.get_system_prompt_from_kwargs(kwargs=None) is None
def test_get_standard_logging_object_payload_includes_system_prompt_for_list_system(logging_obj):
"""List-form Anthropic system blocks must appear on the payload without mutating messages."""
import datetime
from litellm.litellm_core_utils.litellm_logging import get_standard_logging_object_payload

View file

@ -14,6 +14,7 @@ from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.redact_messages import (
_redact_responses_api_output,
perform_redaction,
redact_message_input_output_from_logging,
redact_streaming_responses_for_custom_logger,
should_redact_message_logging,
)
@ -181,6 +182,9 @@ class TestPerformRedaction:
"input": "sensitive input",
"standard_logging_object": {
"messages": [{"role": "user", "content": "sensitive input"}],
"system_prompt": [
{"type": "text", "text": "SHAPE-SECRET", "cache_control": {"type": "ephemeral"}},
],
"response": {
"output": [
{"text": "top-level text"},
@ -207,6 +211,7 @@ class TestPerformRedaction:
]
assert details["prompt"] == ""
assert details["input"] == ""
assert details["standard_logging_object"]["system_prompt"] == "redacted-by-litellm"
logged_response = details["standard_logging_object"]["response"]
assert logged_response["usage"] == {"total_tokens": 1}
@ -720,3 +725,47 @@ class TestRedactStreamingResponsesForCustomLogger:
assert result_details is model_call_details
assert response_obj.choices[0].message.content == "secret content"
class TestSystemPromptRedaction:
REDACTED = "redacted-by-litellm"
SYSTEM_PROMPT = [{"type": "text", "text": "SHAPE-SECRET"}]
def _details_with_system_prompt(self, **extra):
details = {
"messages": [{"role": "user", "content": "hi"}],
"standard_logging_object": {
"messages": [{"role": "user", "content": "hi"}],
"system_prompt": self.SYSTEM_PROMPT,
"response": {"choices": [{"message": {"content": "secret"}}]},
},
"litellm_params": {"metadata": {}},
}
details.update(extra)
return details
def test_global_turn_off_message_logging_redacts_system_prompt(self):
litellm.turn_off_message_logging = True
details = self._details_with_system_prompt()
redact_message_input_output_from_logging(details, result=None)
assert details["standard_logging_object"]["system_prompt"] == self.REDACTED
def test_request_level_dynamic_param_redacts_system_prompt(self):
details = self._details_with_system_prompt(
standard_callback_dynamic_params={"turn_off_message_logging": True}
)
redact_message_input_output_from_logging(details, result=None)
assert details["standard_logging_object"]["system_prompt"] == self.REDACTED
def test_per_callback_turn_off_message_logging_redacts_system_prompt(self):
details = self._details_with_system_prompt()
logger = CustomLogger(turn_off_message_logging=True)
result = logger.redact_standard_logging_payload_from_model_call_details(details)
assert result["standard_logging_object"]["system_prompt"] == self.REDACTED
assert details["standard_logging_object"]["system_prompt"] == self.SYSTEM_PROMPT