fix(guardrails): record not_run when a skipped role mixes text and images

Only image-only unscoped content stays unrecorded; text or tool content
removed by scoping is recorded as not_run even when an image sits beside it.
Also keeps the type-discipline budget flat by returning the reason from the
helper and annotating the accumulator lists _extract_inputs requires.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-15 05:20:56 +00:00
parent 714b113c5f
commit fe0fb97fd2
3 changed files with 52 additions and 27 deletions

View file

@ -213,8 +213,16 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
task_mappings=tool_call_task_mappings,
)
elif not images_to_check and not guardrail_to_apply.records_own_guardrail_information:
self._record_not_run(data=data, messages=messages, guardrail_to_apply=guardrail_to_apply)
elif (
not images_to_check
and not guardrail_to_apply.records_own_guardrail_information
and (not_run_reason := self._not_run_reason(messages)) is not None
):
guardrail_to_apply.add_standard_logging_guardrail_information_to_request_data(
guardrail_json_response=not_run_reason,
request_data=data,
guardrail_status="not_run",
)
verbose_proxy_logger.debug(
"OpenAI Chat Completions: Processed input messages: %s",
@ -223,36 +231,27 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
return data
def _record_not_run(
def _not_run_reason(
self,
data: dict,
messages: list[dict[str, Any]],
guardrail_to_apply: "CustomGuardrail",
) -> None:
unscoped_texts: Final[list[str]] = []
unscoped_images: Final[list[str]] = []
unscoped_tool_calls: Final[list[ChatCompletionToolParam]] = []
messages: Sequence[dict[str, Any]], # mutable-ok: raw request messages consumed by _extract_inputs
) -> str | None:
"""Why nothing was scanned, or None when the only unscoped content is images, which this handler never scans."""
texts: Final[list[str]] = [] # mutable-ok: filled by _extract_inputs
images: Final[list[str]] = [] # mutable-ok: filled by _extract_inputs
tool_calls: Final[list[ChatCompletionToolParam]] = [] # mutable-ok: filled by _extract_inputs
for msg_idx, message in enumerate(messages):
self._extract_inputs(
message=message,
msg_idx=msg_idx,
texts_to_check=unscoped_texts,
images_to_check=unscoped_images,
tool_calls_to_check=unscoped_tool_calls,
text_task_mappings=[],
tool_call_task_mappings=[],
texts_to_check=texts,
images_to_check=images,
tool_calls_to_check=tool_calls,
text_task_mappings=[], # mutable-ok: required by _extract_inputs, unused here
tool_call_task_mappings=[], # mutable-ok: required by _extract_inputs, unused here
)
if unscoped_images:
return
guardrail_to_apply.add_standard_logging_guardrail_information_to_request_data(
guardrail_json_response=(
"no scannable content after message scoping"
if unscoped_texts or unscoped_tool_calls
else "no scannable content"
),
request_data=data,
guardrail_status="not_run",
)
if texts or tool_calls:
return "no scannable content after message scoping"
return None if images else "no scannable content"
def extract_request_tool_names(self, data: dict) -> list[str]:
"""Extract tool names from OpenAI chat completions request (tools[].function.name, functions[].name)."""

View file

@ -26,7 +26,7 @@ class ComplianceChecker:
def __init__(self, data: ComplianceCheckRequest):
self.data = data
self.guardrails = [g for g in (data.guardrail_information or []) if g.get("guardrail_status") != "not_run"]
self.guardrails = tuple(g for g in data.guardrail_information or () if g.get("guardrail_status") != "not_run")
def _get_guardrails_by_mode(self, mode: str) -> list[dict]:
"""

View file

@ -2000,6 +2000,32 @@ class TestNoScannableContentRecordsNotRun:
assert guardrail.last_inputs is None
assert self._recorded_entries(data) == []
@pytest.mark.asyncio
async def test_scoped_out_text_with_image_records_not_run(self):
"""Scoping removed text too, so the skip is recorded even though an image sat beside it"""
handler = OpenAIChatCompletionsHandler()
guardrail = MockGuardrail(guardrail_name="image-guardrail")
guardrail.skip_system_message_in_guardrail = True
data = {
"messages": [
{
"role": "system",
"content": [
{"type": "text", "text": "Describe this picture."},
{"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}},
],
},
]
}
await handler.process_input_messages(data=data, guardrail_to_apply=guardrail)
assert guardrail.last_inputs is None
entries = self._recorded_entries(data)
assert len(entries) == 1
assert entries[0]["guardrail_status"] == "not_run"
assert entries[0]["guardrail_response"] == "no scannable content after message scoping"
class ToolDroppingTextGuardrail(CustomGuardrail):
"""Answers one text per non-tool message it saw, the way a guardrail that