fix(guardrails): harden response-scan conversation after review

Restore the or-{} guard on dynamic request params that an earlier commit
on this branch dropped: a client-sent null extra_body crashed payload
construction, which fail_open turned into a silently unscanned request.

Build no conversation when nothing was extracted from the response, so
conversation-preferring guardrails keep their tool_calls/texts fallback
instead of rescanning the bare request. Normalize assistant-turn tool
calls everywhere (drop the extraction-only index field, accept dict
shapes). Gate qualifire and hiddenlayer v1 so their response scans keep
today's texts semantics, and make the Responses API response scan mirror
its unscoped request scan.

Deduplicate the five attach blocks into one base helper and share the
Anthropic scoping pipeline between request and response scans so the two
cannot drift, translating tools without re-walking the conversation.
This commit is contained in:
Youlian Simidjiyski 2026-08-22 10:44:40 -04:00
parent 9c81b826a0
commit cc71f33fad
13 changed files with 299 additions and 134 deletions

View file

@ -44,10 +44,8 @@ from litellm.types.llms.anthropic import (
)
from litellm.types.llms.openai import (
AllMessageValues,
ChatCompletionAssistantToolCall,
ChatCompletionRequest,
ChatCompletionToolCallChunk,
ChatCompletionToolCallFunctionChunk,
ChatCompletionToolParam,
)
from litellm.types.utils import (
@ -113,6 +111,17 @@ class ExtractedInput:
EMPTY_EXTRACTED_INPUT: Final = ExtractedInput(scanned=(), images=())
@dataclass(frozen=True, slots=True)
class _ScopedRequestView:
"""The request as guardrail scans see it, with the operator scoping applied."""
translated_request: ChatCompletionRequest
full_messages: tuple[AllMessageValues, ...]
scoped_indices: tuple[int, ...]
hoisted_system_message: AllMessageValues | None
has_midturn_system_message: bool
class AnthropicMessagesHandler(BaseTranslation):
"""Process Anthropic messages with guardrails.
@ -316,36 +325,57 @@ class AnthropicMessagesHandler(BaseTranslation):
)
return result if result else None
def _scoped_request_view(
self,
data: dict, # mutable-ok: API request payload
guardrail_to_apply: "CustomGuardrail",
) -> "_ScopedRequestView":
"""
Translate the request the way guardrail scans see it: exclude the
trusted top-level prompt, hoist it back unless skip_system, keep
in-sequence system entries in scope. Single source of truth for the
request and response scans, which must not disagree.
"""
skip_system: Final = effective_skip_system_message_for_guardrail(guardrail_to_apply)
translation_source: Final = { # mutable-ok: API message payload
key: value for key, value in data.items() if key != "system"
}
translated_request: Final = self._translate_to_openai(translation_source)
translated_messages: Final = tuple(translated_request.get("messages") or ())
has_midturn_system_message: Final = any(
str(message.get("role") or "").lower() == "system" for message in translated_messages
)
hoisted_system_message: Final = None if skip_system else self._hoisted_top_level_system_message(data)
full_messages: Final = (
(hoisted_system_message, *translated_messages)
if hoisted_system_message is not None
else translated_messages
)
# skip_system already excluded the trusted top-level prompt (it is simply
# not hoisted); in-sequence system entries are untrusted and stay in scope.
scoped_indices: Final = scoped_structured_message_indices(
full_messages,
scan_only_tool_results=effective_scan_only_tool_results_for_guardrail(guardrail_to_apply),
skip_system=False,
skip_tool=effective_skip_tool_message_for_guardrail(guardrail_to_apply),
)
return _ScopedRequestView(
translated_request=translated_request,
full_messages=full_messages,
scoped_indices=scoped_indices,
hoisted_system_message=hoisted_system_message,
has_midturn_system_message=has_midturn_system_message,
)
def scoped_request_conversation(
self,
request_data: dict, # mutable-ok: API request payload
guardrail_to_apply: "CustomGuardrail",
) -> tuple[AllMessageValues, ...] | None:
"""
Mirror the request scan's scoping: translate without the trusted
top-level prompt, hoist it back unless skip_system, and keep
in-sequence system entries in scope.
"""
if request_data.get("messages") is None:
return None
skip_system: Final = effective_skip_system_message_for_guardrail(guardrail_to_apply)
translation_source: Final = { # mutable-ok: API message payload
key: value for key, value in request_data.items() if key != "system"
}
translated_messages: Final = self._translate_to_openai(translation_source).get("messages")
hoisted_system_message: Final = None if skip_system else self._hoisted_top_level_system_message(request_data)
full_structured_messages: Final = (
(hoisted_system_message, *(translated_messages or ()))
if hoisted_system_message is not None
else tuple(translated_messages or ())
)
scoped_indices: Final = scoped_structured_message_indices(
full_structured_messages,
scan_only_tool_results=effective_scan_only_tool_results_for_guardrail(guardrail_to_apply),
skip_system=False,
skip_tool=effective_skip_tool_message_for_guardrail(guardrail_to_apply),
)
return tuple(full_structured_messages[index] for index in scoped_indices) or None
view: Final = self._scoped_request_view(request_data, guardrail_to_apply)
return tuple(view.full_messages[index] for index in view.scoped_indices) or None
def request_tools_for_guardrail(
self,
@ -356,10 +386,14 @@ class AnthropicMessagesHandler(BaseTranslation):
return None
if not request_data.get("tools"):
return None
translation_source: Final = { # mutable-ok: API message payload
key: value for key, value in request_data.items() if key != "system"
}
tools: Final = self._translate_to_openai(translation_source).get("tools")
probe: Final = self._translate_to_openai(
{ # mutable-ok: API message payload
"model": request_data.get("model") or "",
"messages": [], # mutable-ok: API message payload
"tools": request_data["tools"],
}
)
tools: Final = probe.get("tools")
return tuple(tools) if tools else None
async def process_input_messages(
@ -382,33 +416,17 @@ class AnthropicMessagesHandler(BaseTranslation):
# Exclude only the trusted top-level prompt. In-sequence system entries are untrusted
# and must stay aligned with texts_to_check for positional masking. When the top-level
# prompt is included, the pre-existing count mismatch disables positional masking.
translation_source: Final = { # mutable-ok: API message payload
key: value for key, value in data.items() if key != "system"
}
chat_completion_compatible_request: Final = self._translate_to_openai(translation_source)
view: Final = self._scoped_request_view(data, guardrail_to_apply)
full_structured_messages: Final = view.full_messages
has_midturn_system_message: Final = view.has_midturn_system_message
hoisted_system_message: Final = view.hoisted_system_message
scoped_message_indices: Final = view.scoped_indices
structured_messages: Final = [ # mutable-ok: GenericGuardrailAPIInputs takes list
full_structured_messages[index] for index in scoped_message_indices
]
full_structured_messages: Final = cast(
list[AllMessageValues],
chat_completion_compatible_request.get("messages", []),
)
has_midturn_system_message: Final = any(
str(message.get("role") or "").lower() == "system" for message in full_structured_messages
)
hoisted_system_message: Final = None if skip_system else self._hoisted_top_level_system_message(data)
if hoisted_system_message is not None:
full_structured_messages.insert(0, hoisted_system_message)
# skip_system already excluded the trusted top-level prompt (it is simply not hoisted);
# in-sequence system entries are untrusted and always stay in scope.
scoped_message_indices: Final = scoped_structured_message_indices(
full_structured_messages,
scan_only_tool_results=scan_only_tool_results,
skip_system=False,
skip_tool=skip_tool,
)
structured_messages: Final = [full_structured_messages[index] for index in scoped_message_indices]
tools_to_check: Final[list[ChatCompletionToolParam]] = (
[] if scan_only_tool_results else chat_completion_compatible_request.get("tools", [])
tools_to_check: Final[list[ChatCompletionToolParam]] = ( # mutable-ok: GenericGuardrailAPIInputs takes list
[] if scan_only_tool_results else list(view.translated_request.get("tools") or ())
)
# Step 1: Extract all text content and images
@ -944,18 +962,12 @@ class AnthropicMessagesHandler(BaseTranslation):
response,
)
structured_conversation: Final = self.response_scan_conversation(
self.attach_response_scan_context(
inputs,
request_data,
guardrail_to_apply,
self.assistant_turn_from_extraction(texts_to_check, tool_calls_to_check),
)
if structured_conversation:
inputs["structured_messages"] = list(
structured_conversation
) # mutable-ok: GenericGuardrailAPIInputs takes list
response_scan_tools: Final = self.request_tools_for_guardrail(request_data, guardrail_to_apply)
if response_scan_tools:
inputs["tools"] = list(response_scan_tools) # mutable-ok: GenericGuardrailAPIInputs takes list
guardrailed_inputs: Final = await guardrail_to_apply.apply_guardrail(
inputs=inputs,
@ -1024,17 +1036,15 @@ class AnthropicMessagesHandler(BaseTranslation):
key="response",
)
stream_tool_calls: Final = tuple(
ChatCompletionAssistantToolCall(
id=tool_call.id,
type="function",
function=ChatCompletionToolCallFunctionChunk(
name=tool_call.function.name,
arguments=tool_call.function.arguments,
),
self.assistant_tool_call(
tool_call_id=tool_call.id,
name=tool_call.function.name,
arguments=tool_call.function.arguments,
)
for tool_call in tool_calls_list or ()
)
structured_conversation: Final = self.response_scan_conversation(
self.attach_response_scan_context(
guardrail_inputs,
prepared_request_data,
guardrail_to_apply,
self.assistant_turn_from_extraction(
@ -1042,17 +1052,6 @@ class AnthropicMessagesHandler(BaseTranslation):
stream_tool_calls,
),
)
if structured_conversation:
guardrail_inputs["structured_messages"] = list(
structured_conversation
) # mutable-ok: GenericGuardrailAPIInputs takes list
response_scan_tools: Final = self.request_tools_for_guardrail(
prepared_request_data, guardrail_to_apply
)
if response_scan_tools:
guardrail_inputs["tools"] = list(
response_scan_tools
) # mutable-ok: GenericGuardrailAPIInputs takes list
_guardrailed_inputs = await guardrail_to_apply.apply_guardrail(
inputs=guardrail_inputs,
request_data=prepared_request_data,

View file

@ -12,6 +12,7 @@ from litellm.llms.base_llm.guardrail_translation.utils import (
from litellm.types.llms.openai import (
ChatCompletionAssistantMessage,
ChatCompletionAssistantToolCall,
ChatCompletionToolCallFunctionChunk,
)
if TYPE_CHECKING:
@ -21,7 +22,12 @@ if TYPE_CHECKING:
)
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolParam
from litellm.types.llms.openai import (
AllMessageValues,
ChatCompletionToolCallChunk,
ChatCompletionToolParam,
)
from litellm.types.utils import GenericGuardrailAPIInputs
@dataclass(slots=True)
@ -203,16 +209,39 @@ class BaseTranslation(ABC):
with the model's response turns appended.
Returns None when the request context is unavailable (SDK/direct-call
path fabricates request_data without messages); guardrails then fall
back to scanning the extracted texts.
path fabricates request_data without messages) or when nothing was
extracted from the response; guardrails then fall back to scanning the
extracted texts and tool calls, which must not be shadowed by a
conversation that lacks a response turn.
"""
if request_data is None:
if request_data is None or not response_turns:
return None
request_conversation: Final = self.scoped_request_conversation(request_data, guardrail_to_apply)
if request_conversation is None:
return None
return (*request_conversation, *response_turns)
def attach_response_scan_context(
self,
inputs: "GenericGuardrailAPIInputs",
request_data: dict | None, # mutable-ok: API request payload
guardrail_to_apply: "CustomGuardrail",
response_turns: Sequence["AllMessageValues"],
) -> None:
"""
Put the response-scan conversation and the request's tools on ``inputs``
when the request context allows building them; no-op otherwise.
"""
structured_conversation: Final = self.response_scan_conversation(
request_data, guardrail_to_apply, response_turns
)
if structured_conversation is None or request_data is None:
return
inputs["structured_messages"] = list(structured_conversation) # rebind-ok: out-param; field type is list
response_scan_tools: Final = self.request_tools_for_guardrail(request_data, guardrail_to_apply)
if response_scan_tools:
inputs["tools"] = list(response_scan_tools) # rebind-ok: out-parameter; the field type is a list
def request_tools_for_guardrail(
self,
request_data: dict, # mutable-ok: API request payload
@ -225,16 +254,38 @@ class BaseTranslation(ABC):
"""
return None
@staticmethod
def assistant_tool_call(
tool_call_id: str | None,
name: str | None,
arguments: str,
) -> ChatCompletionAssistantToolCall:
"""One assistant-message tool call in the OpenAI wire shape."""
return ChatCompletionAssistantToolCall(
id=tool_call_id,
type="function",
function=ChatCompletionToolCallFunctionChunk(name=name, arguments=arguments),
)
@staticmethod
def assistant_turn_from_extraction(
texts: Sequence[str],
tool_calls: Sequence["ChatCompletionAssistantToolCall"] | None = None,
tool_calls: Sequence["ChatCompletionAssistantToolCall | ChatCompletionToolCallChunk"] | None = None,
) -> tuple["ChatCompletionAssistantMessage", ...]:
"""
One OpenAI-shape assistant turn built from the texts and tool calls a
handler's response extraction collected; empty when there is nothing.
Tool calls are normalized to the assistant-message shape, dropping
extraction-only fields such as ``index``.
"""
tool_call_items: Final = tuple(tool_calls or ())
tool_call_items: Final = tuple(
BaseTranslation.assistant_tool_call(
tool_call_id=item.get("id"),
name=item["function"].get("name"),
arguments=item["function"].get("arguments") or "",
)
for item in tool_calls or ()
)
if not texts and not tool_call_items:
return ()
if tool_call_items:

View file

@ -36,15 +36,12 @@ from litellm.main import stream_chunk_builder
from litellm.types.llms.openai import (
AllMessageValues,
ChatCompletionAssistantMessage,
ChatCompletionAssistantToolCall,
ChatCompletionToolCallFunctionChunk,
ChatCompletionToolParam,
)
from litellm.types.proxy.guardrails.guardrail_hooks.generic_guardrail_api import (
coerce_stream_holdback_value,
)
from litellm.types.utils import (
ChatCompletionMessageToolCall,
Choices,
GenericGuardrailAPIInputs,
ModelResponse,
@ -403,16 +400,9 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
if hasattr(response, "model") and response.model:
inputs["model"] = response.model
structured_conversation: Final = self.response_scan_conversation(
request_data, guardrail_to_apply, self._build_response_turns(response)
self.attach_response_scan_context(
inputs, request_data, guardrail_to_apply, self._build_response_turns(response)
)
if structured_conversation:
inputs["structured_messages"] = list(
structured_conversation
) # mutable-ok: GenericGuardrailAPIInputs takes list
response_scan_tools: Final = self.request_tools_for_guardrail(request_data, guardrail_to_apply)
if response_scan_tools:
inputs["tools"] = list(response_scan_tools) # mutable-ok: GenericGuardrailAPIInputs takes list
guardrailed_inputs: Final = await guardrail_to_apply.apply_guardrail(
inputs=inputs,
@ -866,16 +856,13 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
def _choice_assistant_turn(self, choice: Choices) -> ChatCompletionAssistantMessage | None:
tool_calls: Final = tuple(
ChatCompletionAssistantToolCall(
id=tool_call.id,
type="function",
function=ChatCompletionToolCallFunctionChunk(
name=tool_call.function.name,
arguments=tool_call.function.arguments,
),
self.assistant_tool_call(
tool_call_id=converted.get("id"),
name=(converted.get("function") or {}).get("name"),
arguments=(converted.get("function") or {}).get("arguments") or "",
)
for tool_call in choice.message.tool_calls or ()
if isinstance(tool_call, ChatCompletionMessageToolCall)
if (converted := self._convert_tool_call_to_dict(tool_call)) is not None and converted.get("function")
)
content: Final = choice.message.content
if content is None and not tool_calls:

View file

@ -248,6 +248,20 @@ class OpenAIResponsesHandler(BaseTranslation):
) = LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools(tools)
tools_to_check.extend(cast(list[ChatCompletionToolParam], transformed_tools))
def scoped_request_conversation(
self,
request_data: dict, # mutable-ok: API request payload
guardrail_to_apply: "CustomGuardrail",
) -> tuple[AllMessageValues, ...] | None:
"""
This surface's request scan sends its structured messages without the
operator scoping flags (the input path applies none), so the response
scan mirrors that; request and response scans must not disagree about
the conversation.
"""
structured_messages: Final = self.get_structured_messages(request_data)
return tuple(structured_messages) if structured_messages else None
def request_tools_for_guardrail(
self,
request_data: dict, # mutable-ok: API request payload
@ -467,18 +481,12 @@ class OpenAIResponsesHandler(BaseTranslation):
if response_model:
inputs["model"] = response_model
structured_conversation: Final = self.response_scan_conversation(
self.attach_response_scan_context(
inputs,
request_data,
guardrail_to_apply,
self.assistant_turn_from_extraction(texts_to_check, tool_calls_to_check),
)
if structured_conversation:
inputs["structured_messages"] = list(
structured_conversation
) # mutable-ok: GenericGuardrailAPIInputs takes list
response_scan_tools: Final = self.request_tools_for_guardrail(request_data, guardrail_to_apply)
if response_scan_tools:
inputs["tools"] = list(response_scan_tools) # mutable-ok: GenericGuardrailAPIInputs takes list
guardrailed_inputs: Final = await guardrail_to_apply.apply_guardrail(
inputs=inputs,
@ -570,18 +578,12 @@ class OpenAIResponsesHandler(BaseTranslation):
if response_model:
inputs["model"] = response_model
structured_conversation: Final = self.response_scan_conversation(
self.attach_response_scan_context(
inputs,
request_data,
guardrail_to_apply,
self.assistant_turn_from_extraction(texts_to_check, tool_calls_to_check),
)
if structured_conversation:
inputs["structured_messages"] = list(
structured_conversation
) # mutable-ok: GenericGuardrailAPIInputs takes list
response_scan_tools: Final = self.request_tools_for_guardrail(request_data, guardrail_to_apply)
if response_scan_tools:
inputs["tools"] = list(response_scan_tools) # mutable-ok: GenericGuardrailAPIInputs takes list
guardrailed_inputs: Final = await guardrail_to_apply.apply_guardrail(
inputs=inputs,

View file

@ -187,7 +187,10 @@ class GraySwanGuardrail(CustomGuardrail):
input_type,
)
dynamic_body: Final = self.get_guardrail_dynamic_request_body_params(request_data)
dynamic_body: Final = (
self.get_guardrail_dynamic_request_body_params(request_data)
or {} # mutable-ok: empty fallback when the client sent a null extra_body
)
if dynamic_body:
verbose_proxy_logger.debug("Gray Swan Guardrail: dynamic extra_body=%s", safe_dumps(dynamic_body))
@ -521,7 +524,7 @@ class GraySwanGuardrail(CustomGuardrail):
if conversation:
return conversation, self._sanitize_json_list(inputs.get("tools"))
if input_type == "request":
return self._texts_fallback(inputs, "user"), None
return self._texts_fallback(inputs), None
return self._build_response_turns(inputs), None
def _build_response_turns(self, inputs: GenericGuardrailAPIInputs) -> tuple[Mapping[str, Any], ...]:
@ -533,8 +536,8 @@ class GraySwanGuardrail(CustomGuardrail):
final_turn: Final[MonitorTurn] = {**base[-1], "tool_calls": tool_calls}
return (*base[:-1], final_turn)
def _texts_fallback(self, inputs: GenericGuardrailAPIInputs, role: str) -> tuple[MonitorTurn, ...]:
return tuple(self._turn(role, text) for text in inputs.get("texts", ()))
def _texts_fallback(self, inputs: GenericGuardrailAPIInputs) -> tuple[MonitorTurn, ...]:
return tuple(self._turn("user", text) for text in inputs.get("texts", ()))
def _turn(self, role: str, content: str) -> MonitorTurn:
turn: Final[MonitorTurn] = {"role": role, "content": content}

View file

@ -211,7 +211,9 @@ class HiddenlayerGuardrail(CustomGuardrail):
hl_request_metadata["requester_id"] = headers.get("hl-requester-id") or "LiteLLM"
project_id: Final = headers.get("hl-project-id")
if scan_params := inputs.get("structured_messages"):
# Response scans keep the texts path: their structured_messages carry the
# whole conversation, whose last turn is not necessarily the scan target.
if input_type == "request" and (scan_params := inputs.get("structured_messages")):
last_msg: Final = scan_params[-1]
result: _HiddenlayerResponse = await self._call_hiddenlayer(
project_id,

View file

@ -424,8 +424,10 @@ class QualifireGuardrail(CustomGuardrail):
# Get dynamic params from request body (allows runtime overrides)
dynamic_params: Final = self.get_guardrail_dynamic_request_body_params(request_data=request_data)
# Extract messages from structured_messages or request_data
messages: list[AllMessageValues] | None = inputs.get("structured_messages")
# Extract messages from structured_messages or request_data. Response
# scans keep the request_data path: their structured_messages carry the
# model's answer too, which would land in `messages` on top of `output`.
messages: list[AllMessageValues] | None = inputs.get("structured_messages") if input_type == "request" else None
if not messages:
messages = request_data.get("messages")

View file

@ -1830,7 +1830,7 @@ class MockInputsRecordingGuardrail(CustomGuardrail):
def __init__(self):
super().__init__(guardrail_name="inputs-recording")
self.calls: list = []
self.calls: list[tuple[str, dict]] = []
async def apply_guardrail(
self,
@ -1940,6 +1940,9 @@ class TestAnthropicResponseScanConversation:
assert assistant_turn["role"] == "assistant"
assert assistant_turn["content"] == "Checking."
assert assistant_turn["tool_calls"][0]["function"]["name"] == "get_weather"
assert "index" not in assistant_turn["tool_calls"][0], (
"extraction-only fields must not leak into the assistant-message tool call shape"
)
assert inputs["tools"][0]["function"]["name"] == "get_weather"
@pytest.mark.asyncio

View file

@ -1571,7 +1571,7 @@ class RecordingGuardrail(CustomGuardrail):
def __init__(self):
super().__init__(guardrail_name="recording")
self.calls: list = []
self.calls: list[tuple[str, dict]] = []
async def apply_guardrail(
self,
@ -1775,3 +1775,13 @@ class TestResponseScanConversation:
{"role": "user", "content": "say hi"},
{"role": "assistant", "content": "hi there"},
]
@pytest.mark.asyncio
async def test_empty_response_turns_yield_no_conversation(self):
"""A conversation without a response turn would shadow the tool_calls/texts
fallback in conversation-preferring guardrails, so none is built."""
handler = OpenAIChatCompletionsHandler()
guardrail = RecordingGuardrail()
request_data = {"messages": [{"role": "user", "content": "hi"}]}
assert handler.response_scan_conversation(request_data, guardrail, []) is None

View file

@ -1241,7 +1241,7 @@ class MockInputsRecordingGuardrail(CustomGuardrail):
def __init__(self):
super().__init__(guardrail_name="inputs-recording")
self.calls: list = []
self.calls: list[tuple[str, dict]] = []
async def apply_guardrail(
self,

View file

@ -897,3 +897,29 @@ def test_sanitize_json_list_drops_non_dict_items(grayswan_guardrail: GraySwanGua
assert grayswan_guardrail._sanitize_json_list([{"role": "user", "content": "hi"}, "junk", 3]) == (
{"role": "user", "content": "hi"},
)
@pytest.mark.asyncio
async def test_apply_guardrail_scans_despite_null_dynamic_extra_body(
monkeypatch, grayswan_guardrail: GraySwanGuardrail
) -> None:
"""A client-sent null extra_body must not crash payload construction, which
fail_open would turn into a silently skipped scan."""
captured: dict = {}
async def _fake_call(payload: dict):
captured["payload"] = payload
return {"violation": 0.0}
monkeypatch.setattr(grayswan_guardrail, "_call_grayswan_api", _fake_call)
await grayswan_guardrail.apply_guardrail(
inputs={"texts": ["hello"]},
request_data={
"model": "gpt-4",
"metadata": {"guardrails": [{"grayswan-test": {"extra_body": None}}]},
},
input_type="request",
)
assert _wire(captured["payload"])["messages"] == [{"role": "user", "content": "hello"}]

View file

@ -1098,3 +1098,37 @@ class TestHiddenlayerGuardrailV2:
config_model = HiddenlayerGuardrailV2.get_config_model()
assert config_model is not None
assert config_model.__name__ == "HiddenlayerGuardrailConfigModel"
class TestHiddenlayerResponseScanConversation:
@pytest.mark.asyncio
async def test_response_scan_ignores_structured_messages(self, monkeypatch):
"""Response scans now carry the conversation in structured_messages; the
v1 guardrail must keep scanning texts instead of str()-coercing the last
conversation turn (a tool-call-only turn has content None, and scanning
the literal string 'None' would produce junk verdicts)."""
monkeypatch.setenv("HIDDENLAYER_API_BASE", "https://my.hiddenlayer")
guardrail = HiddenlayerGuardrail(
guardrail_name="hiddenlayer", event_hook="post_call", default_on=True
)
async def scan(project_id, metadata, payload, input_type):
if payload["messages"][-1]["content"] == "None":
return {"evaluation": {"action": "BLOCK"}, "analysis": []}
return {}
inputs = GenericGuardrailAPIInputs(
structured_messages=[
{"role": "user", "content": "look this up"},
{"role": "assistant", "content": None},
]
)
with patch.object(guardrail, "_call_hiddenlayer", side_effect=scan):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
input_type="response",
)
assert result is inputs

View file

@ -679,3 +679,49 @@ class TestQualifireGuardrailRegistry:
assert "qualifire" in guardrail_class_registry
assert guardrail_class_registry["qualifire"] == QualifireGuardrail
class TestQualifireResponseScanConversation:
@pytest.mark.asyncio
async def test_response_scan_keeps_request_messages_out_of_conversation(self):
"""Response scans now receive the whole conversation in structured_messages;
Qualifire must keep sending the request messages plus a separate output,
not a conversation that already embeds the model's answer."""
from litellm.proxy.guardrails.guardrail_hooks.qualifire.qualifire import (
QualifireGuardrail,
)
guardrail = QualifireGuardrail(
api_key="test_key",
prompt_injections=True,
guardrail_name="test_guardrail",
)
mock_response = MagicMock()
mock_response.json.return_value = {
"score": 100,
"status": "completed",
"evaluationResults": [],
}
mock_response.raise_for_status = MagicMock()
guardrail.async_handler.post = AsyncMock(return_value=mock_response)
request_messages = [{"role": "user", "content": "what's my balance?"}]
inputs = {
"texts": ["Your balance is $5"],
"structured_messages": request_messages
+ [{"role": "assistant", "content": "Your balance is $5"}],
}
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={"messages": request_messages},
input_type="response",
)
assert result is inputs
payload = guardrail.async_handler.post.call_args[1]["json"]
assert all(m.get("role") != "assistant" for m in payload["messages"]), (
"the model's answer belongs in `output`, not in the conversation"
)
assert payload["output"] == "Your balance is $5"