fix(responses): scan and mask top-level instructions with guardrails (#43629)

* fix(responses): scan and mask top-level instructions with guardrails

The Responses guardrail translation handler put a non-empty top-level instructions field into structured_messages as a system row but never into the flat texts list, so guardrails that scan texts skipped it, flat-text masking could not rewrite it, and PANW latest-only selection failed its alignment guard whenever instructions were present.

Seed texts with the instructions row, carry that offset into the flat-text write-back so a rewritten row lands on data["instructions"], and account for the leading row in the PANW Responses alignment.

Resolves LIT-8931

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(responses): reject empty guardrail rewrites instead of forwarding raw input

An explicit texts=[] answer from a guardrail now fails the count check and
raises UnappliableRequestRewrite like any other misaligned rewrite; only a
missing texts key means no rewrite. Types the out-param as dict[str, object]
and adds integration coverage for instructions blocking, masking, empty
instructions, tool loops, latest-only and concurrent workers

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(responses): type the texts-replacing guardrail helper explicitly

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(responses): honor skip_system_message_in_guardrail for instructions and system input items

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(responses): cover skip_system_message_in_guardrail on the live proxy

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(responses): keep skipped rows through full-coverage rewrites and align latest-only with skip_system

Trust a guardrail's structured_messages_cover_full_request claim only when it
returns as many rows as the full normalized request, otherwise merge the scoped
rows back so skipped instructions and system items survive the write-back.
Make PANW's Responses reasoning alignment skip-aware so latest-only still picks
the latest user turn when system content is excluded from texts.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(responses): annotate new guardrail tests with return types

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(responses): treat an empty guardrail texts answer as no rewrite like chat completions

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(responses): type the guardrail test doubles explicitly

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: yucheng <yucheng@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/llms/openai/responses/guardrail_translation/handler.py

@@ -53,6 +53,10 @@ from litellm.llms.base_llm.guardrail_translation.base_translation import (
 )
 from litellm.llms.base_llm.guardrail_translation.utils import (
     blocked_responses_stream_usage,
+    effective_skip_system_message_for_guardrail,
+    merge_guardrailed_scoped_messages,
+    role_out_of_guardrail_scope,
+    scoped_structured_message_indices,
     stream_item_field,
     stream_item_fingerprint,
     stream_item_items,
@@ -376,6 +380,17 @@ class _RequestFields(NamedTuple):
 class _ExtractedInputs(NamedTuple):
     inputs: GenericGuardrailAPIInputs
     task_mappings: tuple[tuple[int, int | None], ...]
+    instructions: str | None
+
+
+def scannable_instructions(data: Mapping[str, object], *, skip_system: bool = False) -> str | None:
+    instructions: Final = data.get("instructions")
+    return instructions if isinstance(instructions, str) and instructions and not skip_system else None
+
+
+def _input_item_role(item: object) -> str:
+    role: Final = item.get("role") if isinstance(item, Mapping) else None
+    return role.lower() if isinstance(role, str) else ""
 
 
 def _patched_request_fields(
@@ -494,7 +509,14 @@ class OpenAIResponsesHandler(BaseTranslation):
         input_data: Final[str | ResponseInputParam | None] = data.get("input")
         if not isinstance(input_data, (str, list)):
             return data
+        skip_system: Final = effective_skip_system_message_for_guardrail(guardrail_to_apply)
         structured_messages: Final = self.get_structured_messages(data)
+        scoped_indices: Final = scoped_structured_message_indices(
+            structured_messages or [], scan_only_tool_results=False, skip_system=skip_system, skip_tool=False
+        )
+        scoped_structured_messages: Final = (
+            [structured_messages[index] for index in scoped_indices] if structured_messages else None
+        )
         raw_tools: Final = data.get("tools")
         original_tools: Final[tuple[Mapping[str, object], ...]] = (
             tuple(raw_tools) if isinstance(raw_tools, list) else ()
@@ -502,11 +524,13 @@ class OpenAIResponsesHandler(BaseTranslation):
         flattened_tool_groups: Final = tuple(
             form.chat_tools for form in LiteLLMCompletionResponsesConfig.responses_tools_to_chat_forms(original_tools)
         )
-        extracted: Final = self._extract_guardrail_inputs(data, input_data, flattened_tool_groups)
+        extracted: Final = self._extract_guardrail_inputs(
+            data, input_data, flattened_tool_groups, skip_system=skip_system
+        )
         if not extracted.inputs.get("texts"):
             return data
-        if structured_messages:
-            extracted.inputs["structured_messages"] = structured_messages
+        if scoped_structured_messages:
+            extracted.inputs["structured_messages"] = scoped_structured_messages
         guardrailed_inputs: Final = await guardrail_to_apply.apply_guardrail(
             inputs=extracted.inputs,
             request_data=data,
@@ -516,37 +540,63 @@ class OpenAIResponsesHandler(BaseTranslation):
         self._apply_guardrailed_tools_to_data(
             data, original_tools, flattened_tool_groups, guardrailed_inputs.get("tools")
         )
-        written_back: Final = self._written_back_request_fields(data, structured_messages, guardrailed_inputs)
+        written_back: Final = self._written_back_request_fields(
+            data,
+            structured_messages or (),
+            scoped_indices,
+            scoped_structured_messages,
+            guardrail_to_apply,
+            guardrailed_inputs,
+        )
         if written_back is not None:
             data["input"] = list(written_back.input)  # mutable-ok: JSON body
             if written_back.instructions is None:
                 data.pop("instructions", None)
             else:
                 data["instructions"] = written_back.instructions  # rebind-ok: data is an out-param
-        elif isinstance(input_data, str):
-            guardrailed_texts: Final = guardrailed_inputs.get("texts") or ()
-            if len(guardrailed_texts) > 1:
-                raise unappliable_request_rewrite(guardrail_to_apply.guardrail_name)
-            data["input"] = guardrailed_texts[0] if guardrailed_texts else input_data  # rebind-ok: data is an out-param
         else:
-            rewritten_texts: Final = guardrailed_inputs.get("texts") or ()
-            if len(rewritten_texts) != len(extracted.task_mappings):
-                raise unappliable_request_rewrite(guardrail_to_apply.guardrail_name)
-            await self._apply_guardrail_responses_to_input(
-                messages=input_data,
-                responses=rewritten_texts,
-                task_mappings=extracted.task_mappings,
-            )
+            await self._apply_guardrailed_texts(data, input_data, extracted, guardrail_to_apply, guardrailed_inputs)
         verbose_proxy_logger.debug("OpenAI Responses API: Processed input messages: %s", data.get("input"))
         return data
 
+    async def _apply_guardrailed_texts(
+        self,
+        data: dict[str, object],
+        input_data: "str | ResponseInputParam",
+        extracted: _ExtractedInputs,
+        guardrail_to_apply: "CustomGuardrail",
+        guardrailed_inputs: GenericGuardrailAPIInputs,
+    ) -> None:
+        returned_texts: Final = guardrailed_inputs.get("texts")
+        if not returned_texts:
+            return
+        rewritten_texts: Final = tuple(returned_texts)
+        offset: Final = 0 if extracted.instructions is None else 1
+        input_texts: Final = rewritten_texts[offset:]
+        expected: Final = 1 if isinstance(input_data, str) else len(extracted.task_mappings)
+        if len(rewritten_texts) != offset + expected:
+            raise unappliable_request_rewrite(guardrail_to_apply.guardrail_name)
+        if offset:
+            data["instructions"] = rewritten_texts[0]  # rebind-ok: data is an out-param
+        if isinstance(input_data, str):
+            data["input"] = input_texts[0]  # rebind-ok: data is an out-param
+            return
+        await self._apply_guardrail_responses_to_input(
+            messages=input_data,
+            responses=input_texts,
+            task_mappings=extracted.task_mappings,
+        )
+
     def _extract_guardrail_inputs(
         self,
         data: Mapping[str, object],
         input_data: "str | ResponseInputParam",
         flattened_tool_groups: Sequence[Sequence[Mapping[str, object]]],
+        *,
+        skip_system: bool = False,
     ) -> _ExtractedInputs:
-        texts_to_check: Final[list[str]] = []
+        instructions: Final = scannable_instructions(data, skip_system=skip_system)
+        texts_to_check: Final[list[str]] = [] if instructions is None else [instructions]
         images_to_check: Final[list[str]] = []
         task_mappings: Final[list[tuple[int, int | None]]] = []
         tools_to_check: Final[list[ChatCompletionToolParam]] = list(  # mutable-ok: guardrail inputs want a list
@@ -562,6 +612,10 @@ class OpenAIResponsesHandler(BaseTranslation):
             texts_to_check.append(input_data)
         else:
             for msg_idx, message in enumerate(input_data):
+                if role_out_of_guardrail_scope(
+                    _input_item_role(message), skip_system_message=skip_system, skip_tool_message=False
+                ):
+                    continue
                 self._extract_input_text_and_images(
                     message=message,
                     msg_idx=msg_idx,
@@ -577,22 +631,32 @@ class OpenAIResponsesHandler(BaseTranslation):
         model: Final = data.get("model")
         if isinstance(model, str):
             inputs["model"] = model
-        return _ExtractedInputs(inputs=inputs, task_mappings=tuple(task_mappings))
+        return _ExtractedInputs(inputs=inputs, task_mappings=tuple(task_mappings), instructions=instructions)
 
     @staticmethod
     def _written_back_request_fields(
         data: Mapping[str, object],
-        structured_messages: Sequence[AllMessageValues] | None,
+        structured_messages: Sequence[AllMessageValues],
+        scoped_indices: Sequence[int],
+        scoped_structured_messages: Sequence[AllMessageValues] | None,
+        guardrail_to_apply: "CustomGuardrail",
         guardrailed_inputs: GenericGuardrailAPIInputs,
     ) -> _RequestFields | None:
         guardrailed: Final = guardrailed_inputs.get("structured_messages")
-        if guardrailed is None or guardrailed is structured_messages:
+        if guardrailed is None or guardrailed is scoped_structured_messages:
             return None
+        covers_full_request: Final = len(scoped_indices) == len(structured_messages) or (
+            guardrail_to_apply.structured_messages_cover_full_request() and len(guardrailed) == len(structured_messages)
+        )
+        merged: Final = (
+            guardrailed
+            if covers_full_request
+            else merge_guardrailed_scoped_messages(
+                full_messages=structured_messages, scoped_indices=scoped_indices, guardrailed_scoped=guardrailed
+            )
+        )
         return _patch_or_convert_request_fields(
-            data.get("input"),
-            data.get("instructions"),
-            structured_messages or (),
-            guardrailed,
+            data.get("input"), data.get("instructions"), structured_messages, merged
         )
 
     def extract_request_tool_names(self, data: dict) -> list[str]:

litellm/proxy/guardrails/guardrail_hooks/panw_prisma_airs/panw_prisma_airs.py

@@ -28,12 +28,15 @@ from litellm.integrations.custom_guardrail import (
 )
 from litellm.llms.base_llm.guardrail_translation.utils import (
     effective_scan_only_tool_results_for_guardrail,
+    effective_skip_system_message_for_guardrail,
+    role_out_of_guardrail_scope,
 )
 from litellm.llms.custom_httpx.http_handler import (
     AsyncHTTPHandler,
     get_async_httpx_client,
     httpxSpecialProvider,
 )
+from litellm.llms.openai.responses.guardrail_translation.handler import scannable_instructions
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.common_utils.callback_utils import (
     add_guardrail_scan_id,
@@ -105,9 +108,14 @@ class _ResponsesInputItem(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
     type: str | None = None
+    role: str | None = None
     content: str | tuple[_ResponsesContentPart, ...] | None = None
 
-    def text_count(self) -> int:
+    def text_count(self, *, skip_system: bool) -> int:
+        if role_out_of_guardrail_scope(
+            (self.role or "").lower(), skip_system_message=skip_system, skip_tool_message=False
+        ):
+            return 0
         if isinstance(self.content, str):
             return 1
         if self.content is None:
@@ -1636,10 +1644,10 @@ class PanwPrismaAirsHandler(CustomGuardrail):
 
         A message's texts are consumed only when they sit at the running position of
         ``texts``; messages the translation handler added without a counterpart in
-        ``texts`` (Responses ``instructions``, ``function_call_output``, ``reasoning``)
-        are skipped. The walk runs front-to-back and back-to-front and both must agree,
-        so an added message whose text happens to equal a neighbouring real message's
-        text cannot steal that text's attribution. Returns None otherwise.
+        ``texts`` (Responses ``function_call_output``, ``reasoning``) are skipped. The walk
+        runs front-to-back and back-to-front and both must agree, so an added message whose
+        text happens to equal a neighbouring real message's text cannot steal that text's
+        attribution. Returns None otherwise.
         """
         runs: Final = tuple(cls._message_texts(message) for message in messages)
 
@@ -1660,17 +1668,19 @@ class PanwPrismaAirsHandler(CustomGuardrail):
         )
         return forward if len(forward) == len(texts) and forward == backward else None
 
-    @classmethod
+    @staticmethod
     def _reasoning_item_text_indices(
-        cls,
         texts: Sequence[str],
         request_data: Mapping[str, object],
+        *,
+        skip_system: bool,
     ) -> frozenset[int] | None:
         """Return the ``texts`` indices flattened from Responses ``reasoning`` input items.
 
         The Responses translation handler gives those model-authored items the default
         ``user`` role, so the latest-turn selection must not mistake one for a human turn.
         Empty for requests without a Responses ``input`` item list; None when the raw items
+        (after the leading ``instructions`` text, both minus whatever ``skip_system`` drops)
         do not account for every entry of ``texts``.
         """
         try:
@@ -1679,10 +1689,11 @@ class PanwPrismaAirsHandler(CustomGuardrail):
             return None
         if not isinstance(raw_input, tuple):
             return frozenset()
-        counts: Final = tuple(item.text_count() for item in raw_input)
-        if sum(counts) != len(texts):
+        offset: Final = 0 if scannable_instructions(request_data, skip_system=skip_system) is None else 1
+        counts: Final = tuple(item.text_count(skip_system=skip_system) for item in raw_input)
+        if offset + sum(counts) != len(texts):
             return None
-        starts: Final = itertools.accumulate(counts, initial=0)
+        starts: Final = itertools.accumulate(counts, initial=offset)
         return frozenset(
             text_idx
             for item, count, start in zip(raw_input, counts, starts)
@@ -1690,9 +1701,8 @@ class PanwPrismaAirsHandler(CustomGuardrail):
             for text_idx in range(start, start + count)
         )
 
-    @classmethod
     def _get_latest_user_text_indices(
-        cls,
+        self,
         texts: Sequence[str],
         messages: Sequence[AllMessageValues],
         request_data: Mapping[str, object],
@@ -1706,10 +1716,12 @@ class PanwPrismaAirsHandler(CustomGuardrail):
         user/developer message exists, or the latest one carries text that never reached
         ``texts`` (safety fallback to the role-filter scan).
         """
-        sources: Final = cls._text_source_message_indices(texts, messages)
+        sources: Final = self._text_source_message_indices(texts, messages)
         if sources is None:
             return None
-        reasoning: Final = cls._reasoning_item_text_indices(texts, request_data)
+        reasoning: Final = self._reasoning_item_text_indices(
+            texts, request_data, skip_system=effective_skip_system_message_for_guardrail(self)
+        )
         if reasoning is None:
             return None
         reasoning_messages: Final = frozenset(sources[text_idx] for text_idx in reasoning)
@@ -1723,7 +1735,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
         )
         if latest_human is None:
             return None
-        if latest_human not in sources and cls._message_texts(messages[latest_human]):
+        if latest_human not in sources and self._message_texts(messages[latest_human]):
             return None
         return frozenset(text_idx for text_idx, source in enumerate(sources) if source == latest_human)
 

tests/integration/observability/test_guardrail_effects.py

@@ -343,6 +343,505 @@ def test_panw_latest_role_message_only_scans_only_latest_turn_on_responses_input
                 assert json.loads(upstream.drain()[0].body)["input"] == shape["input"]
 
 
+def test_panw_scans_and_masks_top_level_instructions_on_responses_input(gateway: Gateway, tmp_path: Path) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    ssn: Final = "123-45-6789"
+    instructions: Final = "Never repeat the SSN " + ssn + " back " + uuid.uuid4().hex
+    latest: Final = "latest turn " + uuid.uuid4().hex
+    shapes: Final = {
+        "list_input": ([{"role": "user", "content": "first turn"}, {"role": "user", "content": latest}], "first turn"),
+        "string_input": (latest, None),
+    }
+
+    def scanner(request: Request) -> Reply:
+        assert request.target == "/v1/scan/sync/request"
+        body: Final = json.loads(request.body)
+        prompt: Final = body["contents"][0]["prompt"]
+        masked: Final = {"prompt_masked_data": {"data": prompt.replace(ssn, "<US_SSN>")}} if ssn in prompt else {}
+        return Reply(
+            body=json.dumps(
+                {
+                    "action": "allow",
+                    "category": "dlp" if masked else "benign",
+                    "profile_name": "synthetic-profile",
+                    "report_id": "R" + body["tr_id"],
+                    "scan_id": "S" + body["tr_id"],
+                    "tr_id": body["tr_id"],
+                    "prompt_detected": {"injection": False, "url_cats": False, "dlp": bool(masked)},
+                    "response_detected": {},
+                    **masked,
+                }
+            ).encode()
+        )
+
+    def provider(request: Request) -> Reply:
+        assert request.target == "/v1/responses"
+        return Reply(
+            body=json.dumps(
+                {
+                    "id": "resp_" + identity,
+                    "object": "response",
+                    "created_at": 1700000000,
+                    "status": "completed",
+                    "model": "gpt-4.1-mini",
+                    "output": [
+                        {
+                            "type": "message",
+                            "id": "msg_" + identity,
+                            "status": "completed",
+                            "role": "assistant",
+                            "content": [{"type": "output_text", "text": "permitted response", "annotations": []}],
+                        }
+                    ],
+                    "usage": {"input_tokens": 11, "output_tokens": 4, "total_tokens": 15},
+                }
+            ).encode()
+        )
+
+    with wire_server(scanner) as policy, wire_server(provider) as upstream:
+        config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+        config["guardrails"] = [
+            {
+                "guardrail_name": identity,
+                "litellm_params": {
+                    "guardrail": "panw_prisma_airs",
+                    "mode": "pre_call",
+                    "default_on": True,
+                    "api_base": policy.url,
+                    "api_key": "synthetic-panw-key",
+                    "profile_name": "synthetic-profile",
+                },
+            }
+        ]
+        path: Final = tmp_path / "panw.yaml"
+        path.write_text(yaml.safe_dump(config))
+        with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario:
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            for name, (shape, first_turn) in shapes.items():
+                response = candidate.request(
+                    "POST", "/v1/responses", {"model": model, "instructions": instructions, "input": shape}
+                )
+                assert response.status_code == 200, response.text
+                assert response.json()["output"][0]["content"][0]["text"] == "permitted response"
+                scanned = [json.loads(scan.body)["contents"][0]["prompt"] for scan in policy.drain()]
+                expected = [instructions, *([first_turn] if first_turn else []), latest]
+                assert scanned == expected, f"{name}: scanned {scanned}"
+                sent = json.loads(upstream.drain()[0].body)
+                assert sent["instructions"] == instructions.replace(ssn, "<US_SSN>"), f"{name}: sent {sent}"
+                assert sent["input"] == shape, f"{name}: sent {sent}"
+
+
+_SSN: Final = "123-45-6789"
+_MASKED_SSN: Final = "<US_SSN>"
+_DENIED_TERM: Final = "RIGBLOCKME"
+
+
+def _panw_scanner(request: Request) -> Reply:
+    assert request.target == "/v1/scan/sync/request"
+    body: Final = json.loads(request.body)
+    prompt: Final = body["contents"][0]["prompt"]
+    denied: Final = _DENIED_TERM in prompt
+    masked: Final = {"prompt_masked_data": {"data": prompt.replace(_SSN, _MASKED_SSN)}} if _SSN in prompt else {}
+    return Reply(
+        body=json.dumps(
+            {
+                "action": "block" if denied else "allow",
+                "category": "malicious" if denied else ("dlp" if masked else "benign"),
+                "profile_name": "synthetic-profile",
+                "report_id": "R" + body["tr_id"],
+                "scan_id": "S" + body["tr_id"],
+                "tr_id": body["tr_id"],
+                "prompt_detected": {"injection": denied, "url_cats": False, "dlp": bool(masked)},
+                "response_detected": {},
+                **masked,
+            }
+        ).encode()
+    )
+
+
+def _responses_provider(request: Request) -> Reply:
+    if request.method == "GET" and request.target.endswith("/models"):
+        return Reply(body=json.dumps({"object": "list", "data": []}).encode())
+    assert request.target == "/v1/responses", request.target
+    return Reply(
+        body=json.dumps(
+            {
+                "id": "resp_" + uuid.uuid4().hex,
+                "object": "response",
+                "created_at": 1700000000,
+                "status": "completed",
+                "model": "gpt-4.1-mini",
+                "output": [
+                    {
+                        "type": "message",
+                        "id": "msg_synthetic",
+                        "status": "completed",
+                        "role": "assistant",
+                        "content": [{"type": "output_text", "text": "permitted response", "annotations": []}],
+                    }
+                ],
+                "usage": {"input_tokens": 11, "output_tokens": 4, "total_tokens": 15},
+            }
+        ).encode()
+    )
+
+
+def _panw_config(tmp_path: Path, identity: str, policy_url: str, **flags: bool) -> Path:
+    config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    config["guardrails"] = [
+        {
+            "guardrail_name": identity,
+            "litellm_params": {
+                "guardrail": "panw_prisma_airs",
+                "mode": "pre_call",
+                "default_on": True,
+                "api_base": policy_url,
+                "api_key": "synthetic-panw-key",
+                "profile_name": "synthetic-profile",
+                **flags,
+            },
+        }
+    ]
+    path: Final = tmp_path / "panw.yaml"
+    path.write_text(yaml.safe_dump(config))
+    return path
+
+
+def _scanned_prompts(scans: tuple[Request, ...]) -> list[str]:
+    return [json.loads(scan.body)["contents"][0]["prompt"] for scan in scans]
+
+
+def _forwarded_bodies(requests: tuple[Request, ...]) -> list[dict[str, object]]:
+    return [json.loads(request.body) for request in requests if request.method == "POST"]
+
+
+def test_guardrail_denies_responses_request_whose_only_flagged_text_is_in_instructions(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    instructions: Final = "You are terse and say " + _DENIED_TERM + " " + uuid.uuid4().hex
+    shapes: Final = {"string_input": "say hi", "list_input": [{"role": "user", "content": "say hi"}]}
+    with wire_server(_panw_scanner) as policy, wire_server(_responses_provider) as upstream:
+        config: Final = _panw_config(tmp_path, identity, policy.url)
+        with (
+            owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate,
+            candidate.scenario() as scenario,
+        ):
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            for name, shape in shapes.items():
+                response = candidate.request(
+                    "POST", "/v1/responses", {"model": model, "instructions": instructions, "input": shape}
+                )
+                assert response.status_code == 400, f"{name}: {response.text}"
+                assert "Prompt blocked by PANW Prisma AI Security policy" in response.text, response.text
+                assert _scanned_prompts(policy.drain()) == [instructions], name
+                assert _forwarded_bodies(upstream.drain()) == [], (
+                    f"{name}: denied instructions must not reach the provider"
+                )
+
+
+def test_empty_instructions_are_not_scanned_while_input_and_chat_system_masking_are_unchanged(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    secret: Final = "my SSN is " + _SSN + " " + uuid.uuid4().hex
+    masked: Final = secret.replace(_SSN, _MASKED_SSN)
+
+    def chat_provider(request: Request) -> Reply:
+        assert request.target == "/v1/chat/completions"
+        return Reply(
+            body=json.dumps(
+                {
+                    "id": "chatcmpl_" + identity,
+                    "object": "chat.completion",
+                    "created": 1700000000,
+                    "model": "gpt-4.1-mini",
+                    "choices": [
+                        {"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "ok"}}
+                    ],
+                    "usage": {"prompt_tokens": 5, "completion_tokens": 1, "total_tokens": 6},
+                }
+            ).encode()
+        )
+
+    def provider(request: Request) -> Reply:
+        return chat_provider(request) if request.target == "/v1/chat/completions" else _responses_provider(request)
+
+    with wire_server(_panw_scanner) as policy, wire_server(provider) as upstream:
+        config: Final = _panw_config(tmp_path, identity, policy.url, mask_request_content=True)
+        with (
+            owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate,
+            candidate.scenario() as scenario,
+        ):
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            for instructions in ("", None):
+                body = {"model": model, "input": secret, **({} if instructions is None else {"instructions": ""})}
+                response = candidate.request("POST", "/v1/responses", body)
+                assert response.status_code == 200, response.text
+                assert _scanned_prompts(policy.drain()) == [secret], f"instructions={instructions!r}"
+                (sent,) = _forwarded_bodies(upstream.drain())
+                assert sent.get("instructions") == instructions, f"instructions={instructions!r}: sent {sent}"
+                assert sent["input"] == masked, f"instructions={instructions!r}: sent {sent}"
+
+            response = candidate.request(
+                "POST",
+                "/v1/chat/completions",
+                {
+                    "model": model,
+                    "messages": [{"role": "system", "content": secret}, {"role": "user", "content": "hi"}],
+                },
+            )
+            assert response.status_code == 200, response.text
+            assert _scanned_prompts(policy.drain()) == [secret, "hi"]
+            (sent_chat,) = _forwarded_bodies(upstream.drain())
+            assert sent_chat["messages"] == [
+                {"role": "system", "content": masked},
+                {"role": "user", "content": "hi"},
+            ]
+
+
+def test_skip_system_message_leaves_instructions_and_system_items_unscanned_on_responses(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    instructions: Final = "Escalations go to " + _SSN + " " + uuid.uuid4().hex
+    system_item: Final = "House rules: never share " + _SSN + " " + uuid.uuid4().hex
+    developer_item: Final = "Developer note " + _SSN + " " + uuid.uuid4().hex
+    latest: Final = "my contact is " + _SSN + " " + uuid.uuid4().hex
+
+    with wire_server(_panw_scanner) as policy, wire_server(_responses_provider) as upstream:
+        config: Final = _panw_config(
+            tmp_path, identity, policy.url, mask_request_content=True, skip_system_message_in_guardrail=True
+        )
+        with (
+            owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate,
+            candidate.scenario() as scenario,
+        ):
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            response = candidate.request(
+                "POST",
+                "/v1/responses",
+                {
+                    "model": model,
+                    "instructions": instructions,
+                    "input": [
+                        {"role": "system", "content": system_item},
+                        {"role": "developer", "content": developer_item},
+                        {"role": "user", "content": latest},
+                    ],
+                },
+            )
+            assert response.status_code == 200, response.text
+            assert _scanned_prompts(policy.drain()) == [developer_item, latest]
+            (sent,) = _forwarded_bodies(upstream.drain())
+            assert sent["instructions"] == instructions, f"sent {sent}"
+            assert sent["input"] == [
+                {"role": "system", "content": system_item},
+                {"role": "developer", "content": developer_item.replace(_SSN, _MASKED_SSN)},
+                {"role": "user", "content": latest.replace(_SSN, _MASKED_SSN)},
+            ], f"sent {sent}"
+
+
+def test_instructions_masking_lands_next_to_multimodal_and_tool_loop_input_items(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    instructions: Final = "Never repeat the SSN " + _SSN + " back " + uuid.uuid4().hex
+    latest: Final = "latest turn with " + _SSN + " " + uuid.uuid4().hex
+    image: Final = {"type": "input_image", "image_url": "https://example.test/receipt.png", "detail": "low"}
+    shapes: Final = {
+        "multimodal": [
+            {"role": "user", "content": [{"type": "input_text", "text": "first turn"}, image]},
+            {"role": "user", "content": [image, {"type": "input_text", "text": latest}]},
+        ],
+        "tool_loop": [
+            {"role": "user", "content": "first turn"},
+            {"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
+            {"type": "function_call_output", "call_id": "call_1", "output": "tool result with " + _SSN},
+            {"role": "user", "content": latest},
+        ],
+    }
+    with wire_server(_panw_scanner) as policy, wire_server(_responses_provider) as upstream:
+        config: Final = _panw_config(tmp_path, identity, policy.url, mask_request_content=True)
+        with (
+            owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate,
+            candidate.scenario() as scenario,
+        ):
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            for name, shape in shapes.items():
+                response = candidate.request(
+                    "POST", "/v1/responses", {"model": model, "instructions": instructions, "input": shape}
+                )
+                assert response.status_code == 200, f"{name}: {response.text}"
+                assert _scanned_prompts(policy.drain()) == [instructions, "first turn", latest], name
+                (sent,) = _forwarded_bodies(upstream.drain())
+                assert sent["instructions"] == instructions.replace(_SSN, _MASKED_SSN), f"{name}: sent {sent}"
+                expected = json.loads(json.dumps(shape).replace(latest, latest.replace(_SSN, _MASKED_SSN)))
+                assert sent["input"] == expected, f"{name}: sent {sent}"
+
+
+def test_panw_latest_only_with_instructions_masks_only_the_latest_turn(gateway: Gateway, tmp_path: Path) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    instructions: Final = "Keep " + _SSN + " confidential " + uuid.uuid4().hex
+    latest: Final = "latest turn with " + _SSN + " " + uuid.uuid4().hex
+    history: Final = ({"role": "user", "content": "first turn"}, {"role": "assistant", "content": "first reply"})
+    shapes: Final = {
+        "plain": [*history, {"role": "user", "content": latest}],
+        "reasoning": [
+            *history,
+            {"type": "reasoning", "id": "rs_1", "summary": [{"type": "summary_text", "text": "thinking"}]},
+            {"role": "user", "content": latest},
+        ],
+    }
+    with wire_server(_panw_scanner) as policy, wire_server(_responses_provider) as upstream:
+        config: Final = _panw_config(
+            tmp_path, identity, policy.url, mask_request_content=True, experimental_use_latest_role_message_only=True
+        )
+        with (
+            owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate,
+            candidate.scenario() as scenario,
+        ):
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            for name, shape in shapes.items():
+                response = candidate.request(
+                    "POST", "/v1/responses", {"model": model, "instructions": instructions, "input": shape}
+                )
+                assert response.status_code == 200, f"{name}: {response.text}"
+                assert _scanned_prompts(policy.drain()) == [latest], name
+                (sent,) = _forwarded_bodies(upstream.drain())
+                assert sent["instructions"] == instructions, f"{name}: latest-only must leave instructions alone"
+                assert sent["input"] == [*shape[:-1], {"role": "user", "content": latest.replace(_SSN, _MASKED_SSN)}], (
+                    f"{name}: sent {sent}"
+                )
+
+
+def test_bedrock_latest_only_masks_latest_turn_on_responses_input_with_instructions(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    guardrail_id: Final = "synthetic" + uuid.uuid4().hex[:8]
+    instructions: Final = "Keep " + _SSN + " confidential " + uuid.uuid4().hex
+    latest: Final = "latest turn with " + _SSN + " " + uuid.uuid4().hex
+
+    def guardrail(request: Request) -> Reply:
+        assert request.target == f"/guardrail/{guardrail_id}/version/DRAFT/apply", request.target
+        body: Final = json.loads(request.body)
+        assert body["source"] == "INPUT", body
+        assert body["content"] == [{"text": {"text": latest}}], body
+        return Reply(
+            body=json.dumps(
+                {
+                    "action": "GUARDRAIL_INTERVENED",
+                    "outputs": [{"text": latest.replace(_SSN, _MASKED_SSN)}],
+                    "assessments": [
+                        {
+                            "sensitiveInformationPolicy": {
+                                "piiEntities": [
+                                    {"type": "US_SOCIAL_SECURITY_NUMBER", "match": _SSN, "action": "ANONYMIZED"}
+                                ]
+                            }
+                        }
+                    ],
+                }
+            ).encode()
+        )
+
+    with wire_server(guardrail) as policy, wire_server(_responses_provider) as upstream:
+        config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+        config["guardrails"] = [
+            {
+                "guardrail_name": identity,
+                "litellm_params": {
+                    "guardrail": "bedrock",
+                    "mode": "pre_call",
+                    "default_on": True,
+                    "mask_request_content": True,
+                    "experimental_use_latest_role_message_only": True,
+                    "guardrailIdentifier": guardrail_id,
+                    "guardrailVersion": "DRAFT",
+                    "aws_region_name": "us-east-1",
+                    "aws_access_key_id": "AKIASYNTHETICGUARDRAIL",
+                    "aws_secret_access_key": "synthetic-secret",
+                    "aws_bedrock_runtime_endpoint": policy.url,
+                },
+            }
+        ]
+        path: Final = tmp_path / "bedrock-instructions.yaml"
+        path.write_text(yaml.safe_dump(config))
+        with owned_proxy(gateway, tmp_path, {}, config=path, workers=2) as candidate, candidate.scenario() as scenario:
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            response: Final = candidate.request(
+                "POST",
+                "/v1/responses",
+                {
+                    "model": model,
+                    "instructions": instructions,
+                    "input": [
+                        {"role": "user", "content": "first turn"},
+                        {"role": "assistant", "content": "first reply"},
+                        {"role": "user", "content": latest},
+                    ],
+                },
+            )
+            assert response.status_code == 200, response.text
+            assert len(policy.drain()) == 1
+            (sent,) = _forwarded_bodies(upstream.drain())
+            assert sent["instructions"] == instructions, sent
+            assert sent["input"] == [
+                {"role": "user", "content": "first turn"},
+                {"role": "assistant", "content": "first reply"},
+                {"role": "user", "content": latest.replace(_SSN, _MASKED_SSN)},
+            ], sent
+
+
+def test_instructions_masking_holds_under_concurrent_load_across_two_workers(gateway: Gateway, tmp_path: Path) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    with wire_server(_panw_scanner) as policy, wire_server(_responses_provider) as upstream:
+        config: Final = _panw_config(tmp_path, identity, policy.url, mask_request_content=True)
+        with (
+            owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate,
+            candidate.scenario() as scenario,
+        ):
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            tags: Final = tuple(uuid.uuid4().hex for _ in range(16))
+
+            def send(tag: str) -> httpx.Response:
+                return candidate.request(
+                    "POST",
+                    "/v1/responses",
+                    {"model": model, "instructions": "Keep " + _SSN + " private " + tag, "input": "say hi " + tag},
+                )
+
+            with ThreadPoolExecutor(max_workers=8) as pool:
+                responses: Final = tuple(pool.map(send, tags))
+            assert [response.status_code for response in responses] == [200] * len(tags), [
+                response.text for response in responses
+            ]
+            sent: Final = {str(body["input"]): body for body in _forwarded_bodies(upstream.drain())}
+            assert sorted(_scanned_prompts(policy.drain())) == sorted(
+                [text for tag in tags for text in ("Keep " + _SSN + " private " + tag, "say hi " + tag)]
+            )
+            assert {tag: sent["say hi " + tag]["instructions"] for tag in tags} == {
+                tag: "Keep " + _MASKED_SSN + " private " + tag for tag in tags
+            }
+
+
 @pytest.mark.covers("other.observability.guardrails.bedrock_passthrough_converse_scans_only_caller_content")
 def test_bedrock_passthrough_converse_guardrail_ignores_denied_term_in_tool_definition(
     gateway: Gateway, tmp_path: Path

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_crowdstrike_aidr.py

@@ -1,7 +1,7 @@
+import json
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from typing import Final, cast
-import json
 from unittest.mock import patch
 
 import httpx
@@ -12,9 +12,9 @@ from pydantic import ValidationError
 import litellm
 from litellm.exceptions import Timeout
 from litellm.integrations.custom_guardrail import CustomGuardrail
+from litellm.litellm_core_utils.core_helpers import get_or_create_metadata_bucket
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
 from litellm.llms.openai.responses.guardrail_translation.handler import OpenAIResponsesHandler
-from litellm.litellm_core_utils.core_helpers import get_or_create_metadata_bucket
 from litellm.proxy.guardrails.guardrail_hooks.crowdstrike_aidr import initialize_guardrail
 from litellm.proxy.guardrails.guardrail_hooks.crowdstrike_aidr.crowdstrike_aidr import (
     CrowdStrikeAIDRGuardrailMissingSecrets,
@@ -1805,36 +1805,22 @@ class _MessageShapedGuardrail(CustomGuardrail):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    ("case", "instructions", "responses_input"),
-    [
-        (
-            "instructions add a system message",
-            "be terse",
-            [{"role": "user", "content": [{"type": "input_text", "text": "my ssn is 078-05-1120"}]}],
-        ),
-        (
-            "tool items add messages that carry no text",
-            None,
-            [
-                {"role": "user", "content": [{"type": "input_text", "text": "my ssn is 078-05-1120"}]},
-                {"type": "function_call", "call_id": "c1", "name": "get_x", "arguments": "{}"},
-                {"type": "function_call_output", "call_id": "c1", "output": "42"},
-            ],
-        ),
-    ],
+    ("case", "instructions"),
+    [("tool items add messages that carry no text", None), ("instructions do not rescue the tool desync", "be terse")],
 )
-async def test_unalignable_rewrite_is_rejected_never_sent_unredacted(
-    case: str,
-    instructions: str | None,
-    responses_input: list[dict[str, object]],
-) -> None:
+async def test_unalignable_rewrite_is_rejected_never_sent_unredacted(case: str, instructions: str | None) -> None:
     """An unalignable rewrite must fail the request, not forward the raw prompt.
 
     Skipping the write-back would hand the model the unredacted text, so a
-    guardrail could be bypassed by adding ``instructions`` or a tool call.
+    guardrail could be bypassed by adding a tool call.
     """
     from litellm.llms.base_llm.guardrail_translation.utils import UnappliableRequestRewrite
 
+    responses_input: list[dict[str, object]] = [
+        {"role": "user", "content": [{"type": "input_text", "text": "my ssn is 078-05-1120"}]},
+        {"type": "function_call", "call_id": "c1", "name": "get_x", "arguments": "{}"},
+        {"type": "function_call_output", "call_id": "c1", "output": "42"},
+    ]
     data: dict[str, object] = {"model": "gpt-4o", "input": responses_input}
     if instructions is not None:
         data["instructions"] = instructions
@@ -1846,21 +1832,27 @@ async def test_unalignable_rewrite_is_rejected_never_sent_unredacted(
         )
 
     assert "078-05-1120" in str(responses_input), case
+    assert data.get("instructions") == instructions, case
 
 
 @pytest.mark.asyncio
-async def test_aligned_rewrite_is_written_back() -> None:
-    """Matching counts must still redact the input in place."""
+@pytest.mark.parametrize("instructions", [None, "be terse"])
+async def test_aligned_rewrite_is_written_back(instructions: str | None) -> None:
+    """Matching counts must redact the input, and the instructions when present, in place."""
     responses_input: list[dict[str, object]] = [
         {"role": "user", "content": [{"type": "input_text", "text": "my ssn is 078-05-1120"}]}
     ]
+    data: dict[str, object] = {"model": "gpt-4o", "input": responses_input}
+    if instructions is not None:
+        data["instructions"] = instructions
 
     await OpenAIResponsesHandler().process_input_messages(
-        data={"model": "gpt-4o", "input": responses_input},
+        data=data,
         guardrail_to_apply=_MessageShapedGuardrail("my ssn is <US_SSN>"),
     )
 
     assert cast(list, responses_input[0]["content"])[0]["text"] == "my ssn is <US_SSN>"
+    assert data.get("instructions") == (None if instructions is None else "my ssn is <US_SSN>")
 
 
 @pytest.mark.asyncio

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_panw_prisma_airs.py

@@ -4867,7 +4867,39 @@ class TestPanwAirsLatestRoleMessageOnlyEveryRequestShape:
         assert result["input"][0]["content"] == "First user turn"
 
     @pytest.mark.asyncio
-    async def test_flag_false_responses_scans_full_history(self):
+    @pytest.mark.parametrize(
+        "history_tail",
+        [
+            pytest.param((), id="plain"),
+            pytest.param(
+                ({"type": "reasoning", "id": "rs_1", "summary": [{"type": "summary_text", "text": "thinking"}]},),
+                id="reasoning",
+            ),
+        ],
+    )
+    async def test_flag_true_with_skip_system_still_scans_only_the_latest_turn_on_responses(
+        self, history_tail: Sequence[Mapping[str, object]]
+    ) -> None:
+        from litellm.llms.openai.responses.guardrail_translation.handler import (
+            OpenAIResponsesHandler,
+        )
+
+        handler = make_handler(experimental_use_latest_role_message_only=True)
+        handler.skip_system_message_in_guardrail = True
+        request_data = self._responses_request(
+            {"role": "system", "content": "House rules"},
+            *history_tail,
+            {"role": "user", "content": self.LATEST},
+            instructions="answer briefly",
+        )
+        patcher, mock_api = self._scan(handler)
+        with patcher:
+            await OpenAIResponsesHandler().process_input_messages(data=request_data, guardrail_to_apply=handler)
+
+        assert [call.kwargs["content"] for call in mock_api.call_args_list] == [self.LATEST]
+
+    @pytest.mark.asyncio
+    async def test_flag_false_responses_scans_instructions_and_full_history(self) -> None:
         from litellm.llms.openai.responses.guardrail_translation.handler import (
             OpenAIResponsesHandler,
         )
@@ -4878,7 +4910,11 @@ class TestPanwAirsLatestRoleMessageOnlyEveryRequestShape:
         with patcher:
             await OpenAIResponsesHandler().process_input_messages(data=request_data, guardrail_to_apply=handler)
 
-        assert [call.kwargs["content"] for call in mock_api.call_args_list] == ["First user turn", self.LATEST]
+        assert [call.kwargs["content"] for call in mock_api.call_args_list] == [
+            "answer briefly",
+            "First user turn",
+            self.LATEST,
+        ]
 
     @pytest.mark.asyncio
     async def test_flag_true_unalignable_texts_fall_back_to_scanning_everything(self):
@@ -4966,8 +5002,12 @@ class TestPanwAirsLatestRoleMessageOnlyEveryRequestShape:
             ),
         ],
     )
+    @pytest.mark.parametrize(
+        "instructions",
+        [pytest.param(None, id="no_instructions"), pytest.param("answer briefly", id="instructions")],
+    )
     async def test_flag_true_reasoning_content_after_latest_user_turn_still_scans_that_turn(
-        self, tail: Sequence[Mapping[str, object]]
+        self, tail: Sequence[Mapping[str, object]], instructions: str | None
     ):
         from litellm.llms.openai.responses.guardrail_translation.handler import (
             OpenAIResponsesHandler,
@@ -4983,6 +5023,7 @@ class TestPanwAirsLatestRoleMessageOnlyEveryRequestShape:
                 "content": [{"type": "reasoning_text", "text": "model chain of thought"}],
             },
             *tail,
+            **({"instructions": instructions} if instructions is not None else {}),
         )
         patcher, mock_api = self._scan(handler)
         with patcher:
@@ -5016,6 +5057,28 @@ class TestPanwAirsLatestRoleMessageOnlyEveryRequestShape:
             "thinking",
         ]
 
+    @pytest.mark.asyncio
+    async def test_flag_true_texts_short_of_the_input_items_fall_back_to_scanning_everything(self) -> None:
+        handler = make_handler(experimental_use_latest_role_message_only=True)
+        reasoning = {"type": "reasoning", "id": "rs_1", "content": [{"type": "reasoning_text", "text": "thinking"}]}
+        inputs: GenericGuardrailAPIInputs = {
+            "texts": ["thinking", self.LATEST],
+            "structured_messages": [{"role": "user", "content": "thinking"}, {"role": "user", "content": self.LATEST}],
+        }
+        request_data: dict[str, object] = {
+            "litellm_call_id": "test-call-id",
+            "input": [
+                {"role": "user", "content": "First user turn"},
+                reasoning,
+                {"role": "user", "content": self.LATEST},
+            ],
+        }
+        patcher, mock_api = self._scan(handler)
+        with patcher:
+            await handler.apply_guardrail(inputs=inputs, request_data=request_data, input_type="request")
+
+        assert [call.kwargs["content"] for call in mock_api.call_args_list] == ["thinking", self.LATEST]
+
 
 class TestPanwAirsMcpToolCallWithoutCallId:
     """Tests for MCP tool invocations flowing through apply_guardrail without

tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py

@@ -7,7 +7,7 @@ with guardrail transformations.
 
 import copy
 from collections.abc import Callable
-from typing import Any, List, Literal, Optional, Tuple
+from typing import Any, Final, List, Literal, Optional, Tuple
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import logging
@@ -67,6 +67,55 @@ class MockGuardrail(CustomGuardrail):
         return inputs
 
 
+class RecordingMaskingGuardrail(MockGuardrail):
+    """MockGuardrail that also records the texts and structured message contents it was shown"""
+
+    def __init__(self, guardrail_name: str) -> None:
+        super().__init__(guardrail_name=guardrail_name)
+        self.seen_texts: list[list[str]] = []
+        self.seen_message_contents: list[list[object]] = []
+
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict[str, object],
+        input_type: Literal["request", "response"],
+        logging_obj: LiteLLMLoggingObj | None = None,
+    ) -> GenericGuardrailAPIInputs:
+        self.seen_texts.append(list(inputs.get("texts", [])))
+        self.seen_message_contents.append([m["content"] for m in inputs.get("structured_messages") or []])
+        return await super().apply_guardrail(inputs, request_data, input_type, logging_obj)
+
+
+class LastTextDroppingGuardrail(CustomGuardrail):
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict[str, object],
+        input_type: Literal["request", "response"],
+        logging_obj: LiteLLMLoggingObj | None = None,
+    ) -> GenericGuardrailAPIInputs:
+        return {**inputs, "texts": list(inputs.get("texts", []))[:-1]}
+
+
+class TextsReplacingGuardrail(CustomGuardrail):
+    """Answers with the given texts list, or without a texts key at all when given None"""
+
+    def __init__(self, guardrail_name: str, texts: tuple[str, ...] | None) -> None:
+        super().__init__(guardrail_name=guardrail_name)
+        self.texts: Final = texts
+
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict[str, object],
+        input_type: Literal["request", "response"],
+        logging_obj: LiteLLMLoggingObj | None = None,
+    ) -> GenericGuardrailAPIInputs:
+        answer: Final = {key: value for key, value in inputs.items() if key != "texts"}
+        return answer if self.texts is None else {**answer, "texts": list(self.texts)}
+
+
 class PersimmonMaskingGuardrail(CustomGuardrail):
     async def apply_guardrail(
         self,
@@ -217,15 +266,9 @@ class TestOpenAIResponsesHandlerInputProcessing:
 
         result = await handler.process_input_messages(data, guardrail)
 
-        assert (
-            result["input"][0]["content"][0]["text"]
-            == "Describe this image [GUARDRAILED]"
-        )
+        assert result["input"][0]["content"][0]["text"] == "Describe this image [GUARDRAILED]"
         # Image URL should remain unchanged
-        assert (
-            result["input"][0]["content"][1]["image_url"]["url"]
-            == "https://example.com/image.jpg"
-        )
+        assert result["input"][0]["content"][1]["image_url"]["url"] == "https://example.com/image.jpg"
 
     @pytest.mark.asyncio
     async def test_process_input_with_empty_content(self):
@@ -248,6 +291,217 @@ class TestOpenAIResponsesHandlerInputProcessing:
         # Empty string should be processed
         assert result["input"][1]["content"] == " [GUARDRAILED]"
 
+    @pytest.mark.asyncio
+    async def test_instructions_over_string_input_are_scanned_first_and_rewritten_in_place(self) -> None:
+        handler = OpenAIResponsesHandler()
+        guardrail = RecordingMaskingGuardrail(guardrail_name="test")
+        data = {"model": "gpt-4", "instructions": "Be terse", "input": "Hello"}
+
+        result = await handler.process_input_messages(data, guardrail)
+
+        assert guardrail.seen_texts == [["Be terse", "Hello"]]
+        assert guardrail.seen_message_contents == [["Be terse", "Hello"]]
+        assert result["instructions"] == "Be terse [GUARDRAILED]"
+        assert result["input"] == "Hello [GUARDRAILED]"
+
+    @pytest.mark.asyncio
+    async def test_instructions_over_list_input_are_scanned_first_and_rewritten_in_place(self) -> None:
+        handler = OpenAIResponsesHandler()
+        guardrail = RecordingMaskingGuardrail(guardrail_name="test")
+        data = {
+            "model": "gpt-4",
+            "instructions": "Be terse",
+            "input": [
+                {"role": "user", "content": "Hello"},
+                {"role": "user", "content": [{"type": "input_text", "text": "World"}]},
+            ],
+        }
+
+        result = await handler.process_input_messages(data, guardrail)
+
+        assert guardrail.seen_texts == [["Be terse", "Hello", "World"]]
+        assert guardrail.seen_message_contents == [["Be terse", "Hello", [{"type": "text", "text": "World"}]]]
+        assert result["instructions"] == "Be terse [GUARDRAILED]"
+        assert result["input"] == [
+            {"role": "user", "content": "Hello [GUARDRAILED]"},
+            {"role": "user", "content": [{"type": "input_text", "text": "World [GUARDRAILED]"}]},
+        ]
+
+    @pytest.mark.asyncio
+    async def test_empty_instructions_are_not_scanned(self) -> None:
+        handler = OpenAIResponsesHandler()
+        guardrail = RecordingMaskingGuardrail(guardrail_name="test")
+        data = {"model": "gpt-4", "instructions": "", "input": "Hello"}
+
+        result = await handler.process_input_messages(data, guardrail)
+
+        assert guardrail.seen_texts == [["Hello"]]
+        assert result["instructions"] == ""
+        assert result["input"] == "Hello [GUARDRAILED]"
+
+    @pytest.mark.asyncio
+    async def test_text_answer_missing_the_instructions_row_is_rejected_and_leaves_request_untouched(self) -> None:
+        from litellm.llms.base_llm.guardrail_translation.utils import UnappliableRequestRewrite
+
+        handler = OpenAIResponsesHandler()
+        guardrail = LastTextDroppingGuardrail(guardrail_name="dropper")
+        data = {"model": "gpt-4", "instructions": "Be terse", "input": [{"role": "user", "content": "Hello"}]}
+        original = copy.deepcopy(data)
+
+        with pytest.raises(UnappliableRequestRewrite) as excinfo:
+            await handler.process_input_messages(data, guardrail)
+
+        assert excinfo.value.guardrail_name == "dropper"
+        assert data["instructions"] == original["instructions"]
+        assert data["input"] == original["input"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("answered_texts", [None, ()], ids=["no_texts_key", "empty_texts"])
+    @pytest.mark.parametrize("data_input", ["Hello", [{"role": "user", "content": "Hello"}]])
+    async def test_answer_without_texts_leaves_instructions_and_input_untouched_like_chat_completions(
+        self, answered_texts: tuple[str, ...] | None, data_input: str | list[dict[str, str]]
+    ) -> None:
+        handler = OpenAIResponsesHandler()
+        guardrail = TextsReplacingGuardrail(guardrail_name="silent", texts=answered_texts)
+        data = {"model": "gpt-4", "instructions": "Be terse", "input": data_input}
+        original = copy.deepcopy(data)
+
+        result = await handler.process_input_messages(data, guardrail)
+
+        assert result["instructions"] == original["instructions"]
+        assert result["input"] == original["input"]
+
+
+def _skipping_system(guardrail: CustomGuardrail) -> CustomGuardrail:
+    guardrail.skip_system_message_in_guardrail = True
+    return guardrail
+
+
+class TestSkipSystemMessageScopesInstructions:
+    """skip_system_message_in_guardrail keeps the Responses system prompt out of the scan the same
+    way it keeps chat `system` messages and Anthropic top-level `system` out: instructions and
+    system-role input items leave both texts and structured_messages, and rewrites leave them verbatim."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("data_input", ["Hello", [{"role": "user", "content": "Hello"}]])
+    async def test_instructions_are_neither_scanned_nor_rewritten(self, data_input: str | list[dict[str, str]]) -> None:
+        handler = OpenAIResponsesHandler()
+        guardrail = _skipping_system(RecordingMaskingGuardrail(guardrail_name="test"))
+        data = {"model": "gpt-4", "instructions": "Be terse", "input": data_input}
+
+        result = await handler.process_input_messages(data, guardrail)
+
+        assert guardrail.seen_texts == [["Hello"]]
+        assert guardrail.seen_message_contents == [["Hello"]]
+        assert result["instructions"] == "Be terse"
+        rewritten = result["input"][0]["content"] if isinstance(data_input, list) else result["input"]
+        assert rewritten == "Hello [GUARDRAILED]"
+
+    @pytest.mark.asyncio
+    async def test_system_input_items_leave_scope_and_user_items_still_align_with_structured_messages(self) -> None:
+        handler = OpenAIResponsesHandler()
+        guardrail = _skipping_system(RecordingMaskingGuardrail(guardrail_name="test"))
+        data = {
+            "model": "gpt-4",
+            "instructions": "Be terse",
+            "input": [
+                {"role": "system", "content": "House rules"},
+                {"role": "developer", "content": "Dev note"},
+                {"role": "user", "content": [{"type": "input_text", "text": "World"}]},
+            ],
+        }
+
+        result = await handler.process_input_messages(data, guardrail)
+
+        assert guardrail.seen_texts == [["Dev note", "World"]]
+        assert guardrail.seen_message_contents == [["Dev note", [{"type": "text", "text": "World"}]]]
+        assert result["instructions"] == "Be terse"
+        assert result["input"] == [
+            {"role": "system", "content": "House rules"},
+            {"role": "developer", "content": "Dev note [GUARDRAILED]"},
+            {"role": "user", "content": [{"type": "input_text", "text": "World [GUARDRAILED]"}]},
+        ]
+
+    @pytest.mark.asyncio
+    async def test_only_system_content_means_nothing_is_scanned(self) -> None:
+        handler = OpenAIResponsesHandler()
+        guardrail = _skipping_system(RecordingMaskingGuardrail(guardrail_name="test"))
+        data = {"model": "gpt-4", "instructions": "Be terse", "input": [{"role": "system", "content": "Rules"}]}
+        original = copy.deepcopy(data)
+
+        result = await handler.process_input_messages(data, guardrail)
+
+        assert guardrail.seen_texts == []
+        assert result == original
+
+    @pytest.mark.asyncio
+    async def test_structured_rewrite_of_the_scoped_rows_keeps_the_skipped_system_prompt(self) -> None:
+        handler = OpenAIResponsesHandler()
+        data = {
+            "model": "gpt-5.6",
+            "instructions": "Answer from the memo only.",
+            "input": [
+                {"role": "system", "content": "House rules"},
+                {"role": "user", "content": "memo " * 400},
+                {"role": "assistant", "content": "Understood."},
+                {"role": "user", "content": "What is the codename?"},
+            ],
+        }
+
+        result = await handler.process_input_messages(data, _skipping_system(StructuredRewriteGuardrail()))
+
+        assert result["instructions"] == "Answer from the memo only."
+        assert [(item["role"], _texts(item)) for item in result["input"]] == [
+            ("system", ["House rules"]),
+            ("user", [COMPRESSED_MARKER]),
+            ("assistant", ["Understood."]),
+            ("user", ["What is the codename?"]),
+        ]
+
+    @pytest.mark.asyncio
+    async def test_full_coverage_claim_over_only_the_scoped_rows_still_keeps_the_skipped_system_prompt(self) -> None:
+        handler = OpenAIResponsesHandler()
+        data = {
+            "model": "gpt-5.6",
+            "instructions": "Answer from the memo only.",
+            "input": [
+                {"role": "system", "content": "House rules"},
+                {"role": "user", "content": "memo " * 400},
+                {"role": "user", "content": "What is the codename?"},
+            ],
+        }
+
+        result = await handler.process_input_messages(data, _skipping_system(ScopedRowsFullCoverageGuardrail()))
+
+        assert result["instructions"] == "Answer from the memo only."
+        assert [(item["role"], _texts(item)) for item in result["input"]] == [
+            ("system", ["House rules"]),
+            ("user", [COMPRESSED_MARKER]),
+            ("user", ["What is the codename?"]),
+        ]
+
+    @pytest.mark.asyncio
+    async def test_full_coverage_claim_over_the_whole_request_is_installed_without_a_second_merge(self) -> None:
+        handler = OpenAIResponsesHandler()
+        data = {
+            "model": "gpt-5.6",
+            "instructions": "Answer from the memo only.",
+            "input": [
+                {"role": "system", "content": "House rules"},
+                {"role": "user", "content": "memo " * 400},
+                {"role": "user", "content": "What is the codename?"},
+            ],
+        }
+
+        result = await handler.process_input_messages(data, _skipping_system(RebuildingFullCoverageGuardrail()))
+
+        assert result["instructions"] == "Answer from the memo only."
+        assert [(item["role"], _texts(item)) for item in result["input"]] == [
+            ("system", ["House rules"]),
+            ("user", [COMPRESSED_MARKER]),
+            ("user", ["What is the codename?"]),
+        ]
+
 
 class TestOpenAIResponsesHandlerOutputProcessing:
     """Test output processing functionality"""
@@ -2156,6 +2410,36 @@ class StructuredRewriteGuardrail(CustomGuardrail):
         return {**inputs, "structured_messages": rewritten}
 
 
+class ScopedRowsFullCoverageGuardrail(StructuredRewriteGuardrail):
+    """Claims its structured_messages span the whole request but, like CrowdStrike AIDR on a
+    Responses body (no `messages` to rebuild from), only ever returns the scoped rows it was given."""
+
+    def structured_messages_cover_full_request(self) -> bool:
+        return True
+
+
+class RebuildingFullCoverageGuardrail(CustomGuardrail):
+    """Claims full coverage and honours it: rebuilds every conversation row from the raw request,
+    compressing the first user turn, the way CrowdStrike AIDR does on a chat body."""
+
+    def structured_messages_cover_full_request(self) -> bool:
+        return True
+
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict[str, object],
+        input_type: Literal["request", "response"],
+        logging_obj: LiteLLMLoggingObj | None = None,
+    ) -> GenericGuardrailAPIInputs:
+        raw_input = request_data["input"]
+        assert isinstance(raw_input, list)
+        full: list[dict[str, object]] = [{"role": "system", "content": request_data["instructions"]}, *raw_input]
+        first_user = next(i for i, m in enumerate(full) if m.get("role") == "user")
+        rewritten = [{**m, "content": COMPRESSED_MARKER} if i == first_user else m for i, m in enumerate(full)]
+        return {**inputs, "structured_messages": rewritten}
+
+
 class ToolOutputRewriteGuardrail(CustomGuardrail):
     """Guardrail that compresses the first tool-result row, the way Headroom does."""
 
@@ -2527,8 +2811,9 @@ def _string_input_request() -> dict:
 class TestPerMessageRewriteWriteBack:
     """A guardrail that rewrites per chat row hands the rows back as
     structured_messages, and the handler lands them on the instructions and the
-    input items they came from; the same rewrite handed back as texts alone has
-    no item to land on and is rejected by name instead of sent unrewritten."""
+    input items they came from; the same rewrite handed back as texts alone lands
+    only where every row has a scanned text (instructions plus a string input) and
+    is otherwise rejected by name instead of sent unrewritten."""
 
     @pytest.mark.asyncio
     async def test_structured_rows_land_on_instructions_and_tool_output(self):
@@ -2576,20 +2861,15 @@ class TestPerMessageRewriteWriteBack:
         assert [_texts(item) for item in result["input"]] == [["My SSN is " + REDACTED_SSN + "."]]
 
     @pytest.mark.asyncio
-    async def test_texts_only_per_message_answer_over_a_string_input_is_rejected_by_name(self):
-        from litellm.llms.base_llm.guardrail_translation.utils import UnappliableRequestRewrite
-
+    async def test_texts_only_per_message_answer_over_a_string_input_lands_on_instructions_and_input(self) -> None:
         guardrail = _per_message_redactor()
         data = _string_input_request()
-        original = copy.deepcopy(data)
 
         with patch.object(guardrail.async_handler, "post", side_effect=_per_message_guardrail_server(False)):
-            with pytest.raises(UnappliableRequestRewrite) as excinfo:
-                await OpenAIResponsesHandler().process_input_messages(data, guardrail)
+            result = await OpenAIResponsesHandler().process_input_messages(data, guardrail)
 
-        assert excinfo.value.guardrail_name == "per-message-redactor"
-        assert data["input"] == original["input"]
-        assert data["instructions"] == original["instructions"]
+        assert result["instructions"] == "Never repeat the SSN " + REDACTED_SSN + " back."
+        assert result["input"] == "My SSN is " + REDACTED_SSN + "."
 
 
 class TestProvenancePatching: