fix(panw_prisma_airs): honor experimental_use_latest_role_message_only on every request shape (#42447)

* fix(panw_prisma_airs): apply experimental_use_latest_role_message_only to every request shape

Explicit true/false now applies to chat completions, Anthropic /v1/messages and /v1/responses alike; unset keeps latest-only for Anthropic and full history otherwise. Text indices are mapped back to their source message by value instead of by count, so Responses instructions, function_call_output and reasoning items no longer derail the alignment and silently rescan the whole history

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

* refactor(panw_prisma_airs): type latest-message helpers against AllMessageValues

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

* fix(panw_prisma_airs): require forward and reverse text attribution to agree

A Responses function_call_output whose text equals the latest user turn could claim that turn's slot in a forward-only walk and demote the latest-only scan to an earlier message. Walk both directions and fall back to the full role-filter scan when they disagree

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

* fix(panw_prisma_airs): pick the latest human turn from messages, not from aligned texts

An image-only latest user turn no longer promotes an earlier user turn into the
latest-only scan; it scans nothing on the request side, as the Anthropic path did
before. A latest user/developer message whose text never reached texts (a trailing
Responses reasoning item) falls back to the role-filter scan instead of narrowing.
Types the test helpers, drops the narrating docstrings and adds regressions for both
shapes

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

* fix(panw_prisma_airs): log when latest-only selection leaves nothing to scan

An image-only latest user turn with experimental_use_latest_role_message_only=true intentionally yields zero scanner calls. Emit a debug line naming the call_id so operators can tell this apart from the guardrail not firing.

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

* fix(panw_prisma_airs): keep Responses reasoning items out of latest-turn selection

The Responses translation handler gives reasoning input items the default user
role, so a reasoning item with text content after the latest prompt was picked
as the latest human turn and the real prompt went unscanned under
experimental_use_latest_role_message_only. Map reasoning items back to their
texts positions from the raw input and exclude them; fall back to the
role-filter scan when the raw items do not account for every text

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

---------

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

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

@@ -5,6 +5,8 @@ Palo Alto Networks Prisma AI Runtime Security (AIRS) Guardrail Integration for L
 Provides real-time threat detection, DLP, URL filtering, content masking, and policy enforcement for AI applications.
 """
 
+import functools
+import itertools
 import json
 import os
 import re
@@ -15,7 +17,7 @@ from urllib.parse import urlparse
 
 import httpx
 from fastapi import HTTPException
-from pydantic import BaseModel, ConfigDict, ValidationError, field_validator
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError, field_validator
 
 from litellm._logging import verbose_proxy_logger
 from litellm._uuid import uuid
@@ -38,6 +40,7 @@ from litellm.proxy.common_utils.callback_utils import (
     add_guardrail_to_applied_guardrails_header,
 )
 from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.llms.openai import AllMessageValues
 from litellm.types.utils import (
     CallTypes,
     CallTypesLiteral,
@@ -90,6 +93,32 @@ class _ToolCallSlice(BaseModel):
     function: _ToolCallFunctionSlice | None = None
 
 
+class _ResponsesContentPart(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    text: str | None = None
+
+
+class _ResponsesInputItem(BaseModel):
+    """The slice of a raw Responses ``input`` item that decides which ``texts`` it flattens to."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    type: str | None = None
+    content: str | tuple[_ResponsesContentPart, ...] | None = None
+
+    def text_count(self) -> int:
+        if isinstance(self.content, str):
+            return 1
+        if self.content is None:
+            return 0
+        return sum(part.text is not None for part in self.content)
+
+
+_ResponsesInput: TypeAlias = str | tuple[_ResponsesInputItem, ...] | None
+_RESPONSES_INPUT: Final[TypeAdapter[_ResponsesInput]] = TypeAdapter(_ResponsesInput)
+
+
 if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
     from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
@@ -194,7 +223,6 @@ class PanwPrismaAirsHandler(CustomGuardrail):
         # internal '<=' comparison and surfaces as a misleading api_error.
         self.timeout = float(timeout) if timeout is not None else 10.0
 
-        # Tri-state: None = not set (default-on for Anthropic), True = explicit on, False = explicit off
         self.experimental_use_latest_role_message_only: bool | None = kwargs.get(
             "experimental_use_latest_role_message_only"
         )
@@ -1578,119 +1606,149 @@ class PanwPrismaAirsHandler(CustomGuardrail):
         request_data: Mapping[str, object],
         logging_obj: Optional["LiteLLMLoggingObj"] = None,
     ) -> bool:
-        """Resolve whether to scan only the latest user message.
+        """Resolve whether to scan only the latest user/developer message.
 
-        - Non-Anthropic requests: always False (existing behavior)
-        - Anthropic requests:
-          - Flag explicitly True/False: respect it
-          - Flag None (not set): default to True
+        - Flag explicitly True/False: respect it for every request shape,
+          matching the bedrock guardrail's semantics for the same flag
+        - Flag None (not set): True for Anthropic /v1/messages requests, False otherwise
         """
-        if not self._is_anthropic_request(request_data, logging_obj):
-            return False
-        if self.experimental_use_latest_role_message_only is None:
-            return True  # Default-on for Anthropic
-        return self.experimental_use_latest_role_message_only
+        if self.experimental_use_latest_role_message_only is not None:
+            return self.experimental_use_latest_role_message_only
+        return self._is_anthropic_request(request_data, logging_obj)
 
     @staticmethod
-    def _get_latest_user_text_indices(
+    def _message_texts(message: AllMessageValues) -> tuple[str, ...]:
+        """Text entries the framework flattens out of one structured message."""
+        content: Final = message.get("content")
+        if isinstance(content, str):
+            return (content,)
+        if not isinstance(content, list):
+            return ()
+        return tuple(text for item in content if isinstance(item, dict) and isinstance(text := item.get("text"), str))
+
+    @classmethod
+    def _text_source_message_indices(
+        cls,
         texts: Sequence[str],
-        messages: Sequence[object],
-    ) -> set | None:
+        messages: Sequence[AllMessageValues],
+    ) -> tuple[int, ...] | None:
+        """Map every ``texts`` entry to the index of the structured message it was flattened from.
+
+        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.
+        """
+        runs: Final = tuple(cls._message_texts(message) for message in messages)
+
+        def walk(ordered_runs: Sequence[tuple[str, ...]], ordered_texts: Sequence[str]) -> tuple[int, ...]:
+            def consume(sources: tuple[int, ...], item: tuple[int, tuple[str, ...]]) -> tuple[int, ...]:
+                position, run = item
+                start: Final = len(sources)
+                if run and tuple(ordered_texts[start : start + len(run)]) == run:
+                    return sources + (position,) * len(run)
+                return sources
+
+            return functools.reduce(consume, enumerate(ordered_runs), ())
+
+        forward: Final = walk(runs, texts)
+        last: Final = len(runs) - 1
+        backward: Final = tuple(
+            last - position for position in walk(tuple(run[::-1] for run in runs[::-1]), texts[::-1])[::-1]
+        )
+        return forward if len(forward) == len(texts) and forward == backward else None
+
+    @classmethod
+    def _reasoning_item_text_indices(
+        cls,
+        texts: Sequence[str],
+        request_data: Mapping[str, object],
+    ) -> 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
+        do not account for every entry of ``texts``.
+        """
+        try:
+            raw_input: Final = _RESPONSES_INPUT.validate_python(request_data.get("input"))
+        except ValidationError:
+            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):
+            return None
+        starts: Final = itertools.accumulate(counts, initial=0)
+        return frozenset(
+            text_idx
+            for item, count, start in zip(raw_input, counts, starts)
+            if item.type == "reasoning"
+            for text_idx in range(start, start + count)
+        )
+
+    @classmethod
+    def _get_latest_user_text_indices(
+        cls,
+        texts: Sequence[str],
+        messages: Sequence[AllMessageValues],
+        request_data: Mapping[str, object],
+    ) -> frozenset[int] | None:
         """Return text indices belonging to only the latest scannable human-authored (user or developer) message.
 
-        Args:
-            texts: Flattened text entries from the framework.
-            messages: The structured messages the framework flattened into ``texts``,
-                      hoisted top-level system prompt included, so positions line up.
-
-        Returns a set of scannable indices, or None on count mismatch or no user/developer
-        message (safety fallback to existing role-filter behavior).
+        The latest user/developer message is chosen from ``messages`` itself, so a latest turn
+        without text (image only) yields an empty set rather than promoting an earlier turn.
+        Messages flattened from Responses ``reasoning`` items are never that turn.
+        Returns None when ``texts`` cannot be aligned with ``messages`` or ``request_data``, no
+        user/developer message exists, or the latest one carries text that never reached
+        ``texts`` (safety fallback to the role-filter scan).
         """
-        last_human_msg_idx: int | None = None
-        for idx in range(len(messages) - 1, -1, -1):
-            msg = messages[idx]
-            if isinstance(msg, dict) and msg.get("role") in ("user", "developer"):
-                last_human_msg_idx = idx
-                break
-
-        if last_human_msg_idx is None:
-            return None  # No user/developer message → fallback to existing role-filter scan
-
-        scannable: Final[set] = set()
-        text_idx = 0
-        for msg_idx, msg in enumerate(messages):
-            if not isinstance(msg, dict):
-                continue
-            content = msg.get("content")
-            is_latest_human = msg_idx == last_human_msg_idx
-
-            if content is None:
-                pass
-            elif isinstance(content, str):
-                if is_latest_human:
-                    scannable.add(text_idx)
-                text_idx += 1
-            elif isinstance(content, list):
-                for item in content:
-                    if isinstance(item, dict) and item.get("text") is not None:
-                        if is_latest_human:
-                            scannable.add(text_idx)
-                        text_idx += 1
-
-        if text_idx != len(texts):
-            return None  # Count mismatch → safety fallback
-
-        return scannable
+        sources: Final = cls._text_source_message_indices(texts, messages)
+        if sources is None:
+            return None
+        reasoning: Final = cls._reasoning_item_text_indices(texts, request_data)
+        if reasoning is None:
+            return None
+        reasoning_messages: Final = frozenset(sources[text_idx] for text_idx in reasoning)
+        latest_human: Final = max(
+            (
+                idx
+                for idx, message in enumerate(messages)
+                if idx not in reasoning_messages and message.get("role") in ("user", "developer")
+            ),
+            default=None,
+        )
+        if latest_human is None:
+            return None
+        if latest_human not in sources and cls._message_texts(messages[latest_human]):
+            return None
+        return frozenset(text_idx for text_idx, source in enumerate(sources) if source == latest_human)
 
     def supports_scan_only_tool_results(self) -> bool:
         return False
 
-    @staticmethod
+    @classmethod
     def _get_scannable_text_indices(
+        cls,
         texts: Sequence[str],
-        structured_messages: Sequence[object],
-    ) -> set | None:
-        """Derive which ``texts`` indices originate from user/system messages.
+        structured_messages: Sequence[AllMessageValues],
+    ) -> frozenset[int] | None:
+        """Derive which ``texts`` indices originate from user/system/developer messages.
 
-        The unified guardrail framework flattens message content into ``texts``
-        without preserving role info.  This helper re-walks
-        ``structured_messages`` using the **same** extraction logic the
-        framework uses (string content → 1 entry, list content → 1 per text
-        item, None → 0) and records the running text index for each entry
-        whose source role is ``"user"``, ``"system"``, or ``"developer"``.
-
-        Returns a set of scannable indices, or ``None`` if the count doesn't
-        match ``len(texts)`` (safety fallback → scan everything).
+        Returns None when ``texts`` cannot be aligned with ``structured_messages``
+        (safety fallback: scan everything).
         """
-        scannable: Final[set] = set()
-        text_idx = 0
-        for msg in structured_messages:
-            if not isinstance(msg, dict):
-                continue
-            role = msg.get("role", "")
-            content = msg.get("content")
-            is_scannable = role in ("user", "system", "developer")
-
-            if content is None:
-                # No content → 0 text entries
-                pass
-            elif isinstance(content, str):
-                if is_scannable:
-                    scannable.add(text_idx)
-                text_idx += 1
-            elif isinstance(content, list):
-                for item in content:
-                    if isinstance(item, dict) and item.get("text") is not None:
-                        if is_scannable:
-                            scannable.add(text_idx)
-                        text_idx += 1
-            # Ignore other content types (shouldn't happen)
-
-        if text_idx != len(texts):
-            # Count mismatch → safety fallback: scan all
+        sources: Final = cls._text_source_message_indices(texts, structured_messages)
+        if sources is None:
             return None
-
-        return scannable
+        return frozenset(
+            text_idx
+            for text_idx, source in enumerate(sources)
+            if structured_messages[source].get("role") in ("user", "system", "developer")
+        )
 
     @staticmethod
     def _mcp_name_fallback(rd: dict) -> str | None:
@@ -1783,16 +1841,18 @@ class PanwPrismaAirsHandler(CustomGuardrail):
 
         # On request side, determine which text indices correspond to scannable
         # messages so we can skip scanning assistant/tool history text.
-        scannable_indices: set | None = None
+        scannable_indices: frozenset[int] | None = None
         if input_type == "request":
             structured_messages: Final = inputs.get("structured_messages")
             if structured_messages:
-                # For Anthropic /v1/messages: default to latest-user-only scanning.
                 if self._use_latest_user_only(request_data, logging_obj):
-                    scannable_indices = self._get_latest_user_text_indices(texts, structured_messages)
-                # Fall through to existing role filtering if:
-                # - not Anthropic, OR flag explicitly False, OR
-                # - latest-user extraction returned None (no user / count mismatch)
+                    scannable_indices = self._get_latest_user_text_indices(texts, structured_messages, request_data)
+                    if scannable_indices is not None and not scannable_indices:
+                        verbose_proxy_logger.debug(
+                            "PANW Prisma AIRS: latest user message has no text, so "
+                            "experimental_use_latest_role_message_only leaves nothing to scan for call_id=%s",
+                            call_id,
+                        )
                 if scannable_indices is None:
                     scannable_indices = self._get_scannable_text_indices(texts, structured_messages)
                 if (

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

@@ -54,9 +54,12 @@ class PanwPrismaAirsGuardrailConfigModel(GuardrailConfigModel):
 
     experimental_use_latest_role_message_only: bool | None = Field(
         default=None,
-        description="Anthropic /v1/messages only. When unset: scans only latest user/developer "
-        "message on request side. Set false to scan all user/system/developer messages. "
-        "Non-Anthropic unaffected.",
+        description="Scan only the latest user/developer message on the request side instead of "
+        "the full conversation history. Set true to enable for every request shape (chat completions, "
+        "Anthropic /v1/messages, /v1/responses); set false to always scan all user/system/developer "
+        "messages. When unset: latest-only for Anthropic /v1/messages, full history otherwise. "
+        "Latest-only trusts caller-supplied history: earlier turns are not rescanned, so enable it only "
+        "where each turn was scanned when it was the latest message or history is server-controlled.",
     )
 
     @staticmethod

tests/integration/observability/test_guardrail_effects.py

@@ -235,6 +235,114 @@ def test_guardrail_denial_prevents_provider_and_preserves_allowed_control(gatewa
             assert len(policy.drain()) == 2
 
 
+def test_panw_latest_role_message_only_scans_only_latest_turn_on_responses_input(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    latest: Final = "latest turn " + uuid.uuid4().hex
+    history: Final = ({"role": "user", "content": "first turn"}, {"role": "assistant", "content": "first reply"})
+    shapes: Final = {
+        "plain": {"input": [*history, {"role": "user", "content": latest}]},
+        "instructions": {"instructions": "answer briefly", "input": [*history, {"role": "user", "content": latest}]},
+        "function_call_output": {
+            "input": [
+                *history,
+                {"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"},
+                {"type": "function_call_output", "call_id": "call_1", "output": "tool result"},
+                {"role": "user", "content": latest},
+            ]
+        },
+        "reasoning": {
+            "input": [
+                *history,
+                {"type": "reasoning", "id": "rs_1", "summary": [{"type": "summary_text", "text": "thinking"}]},
+                {"role": "user", "content": latest},
+            ]
+        },
+        "tool_loop_after_latest": {
+            "input": [
+                *history,
+                {"role": "user", "content": latest},
+                {"type": "reasoning", "id": "rs_2", "content": [{"type": "reasoning_text", "text": "thinking"}]},
+                {"type": "function_call", "call_id": "call_2", "name": "lookup", "arguments": "{}"},
+                {"type": "function_call_output", "call_id": "call_2", "output": "tool result"},
+            ]
+        },
+    }
+
+    def scanner(request: Request) -> Reply:
+        assert request.target == "/v1/scan/sync/request"
+        body: Final = json.loads(request.body)
+        return Reply(
+            body=json.dumps(
+                {
+                    "action": "allow",
+                    "category": "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": False},
+                    "response_detected": {},
+                }
+            ).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",
+                    "experimental_use_latest_role_message_only": True,
+                },
+            }
+        ]
+        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 in shapes.items():
+                response = candidate.request("POST", "/v1/responses", {"model": model, **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()]
+                assert scanned == [latest], f"{name}: latest-only scanned {scanned}"
+                assert json.loads(upstream.drain()[0].body)["input"] == shape["input"]
+
+
 @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