fix(guardrails): record guardrail info in litellm_metadata when client sends metadata

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

litellm/integrations/custom_guardrail.py

@@ -94,6 +94,34 @@ def get_session_id_from_request_data(request_data: Dict[str, Any]) -> Optional[s
     return None
 
 
+def _guardrail_log_containers(request_data: dict) -> tuple:
+    """Return the metadata dicts that guardrail logging info must be written to.
+
+    The spend log for ``/v1/chat/completions`` reads guardrail_information from
+    ``metadata`` while streaming ``anthropic_messages`` reads it from
+    ``litellm_metadata``.  When a client sends a top-level ``metadata`` field
+    (Claude Code / the Anthropic SDK always send ``metadata.user_id``), both
+    containers are present, so the info is recorded into every present container
+    (deduplicated by identity) rather than only the client one.
+    """
+
+    def _ensure_dict(key: str) -> Optional[dict]:
+        if key not in request_data:
+            return None
+        if request_data[key] is None:
+            request_data[key] = {}
+        value = request_data[key]
+        return value if isinstance(value, dict) else None
+
+    containers = tuple(m for m in (_ensure_dict("litellm_metadata"), _ensure_dict("metadata")) if m is not None)
+    if not containers:
+        request_data["metadata"] = {}
+        return (request_data["metadata"],)
+    if len(containers) == 2 and containers[0] is containers[1]:
+        return (containers[0],)
+    return containers
+
+
 class CustomGuardrail(CustomLogger):
     # If True, during_call runs async_moderation_hook instead of the unified apply_guardrail path.
     use_native_during_call_hook: ClassVar[bool] = False
@@ -844,17 +872,8 @@ class CustomGuardrail(CustomLogger):
                 # should not happen
                 container[key] = [existing, slg]
 
-        if "metadata" in request_data:
-            if request_data["metadata"] is None:
-                request_data["metadata"] = {}
-            _append_guardrail_info(request_data["metadata"])
-        elif "litellm_metadata" in request_data:
-            _append_guardrail_info(request_data["litellm_metadata"])
-        else:
-            # Ensure guardrail info is always logged (e.g. proxy may not have set
-            # metadata yet). Attach to "metadata" so spend log / standard logging see it.
-            request_data["metadata"] = {}
-            _append_guardrail_info(request_data["metadata"])
+        for container in _guardrail_log_containers(request_data):
+            _append_guardrail_info(container)
 
         # Emit the otel guardrail span here, where every guardrail execution lands,
         # rather than relying on a post-call hook that does not fire on every path

tests/test_litellm/integrations/test_custom_guardrail.py

@@ -653,6 +653,48 @@ class TestGuardrailLoggingAggregation:
         assert len(info) == 2
         assert info[1]["guardrail_name"] == "test_guardrail"
 
+    def test_records_into_litellm_metadata_when_client_metadata_present(self):
+        """When the client sends a top-level ``metadata`` field (e.g. Claude Code /
+        the Anthropic SDK always send ``metadata.user_id``) alongside the internal
+        ``litellm_metadata``, the guardrail entry must still land in
+        ``litellm_metadata``.  The ``anthropic_messages`` streaming spend log is
+        built from ``litellm_metadata``, so recording only into the client
+        ``metadata`` dict drops guardrail_information and hides the Guardrails
+        section in the UI.
+        """
+        request_data = {
+            "metadata": {"user_id": "client-supplied"},
+            "litellm_metadata": {},
+        }
+
+        self._invoke_add_log(request_data)
+
+        litellm_info = request_data["litellm_metadata"].get(
+            "standard_logging_guardrail_information"
+        )
+        assert isinstance(litellm_info, list)
+        assert len(litellm_info) == 1
+        assert litellm_info[0]["guardrail_name"] == "test_guardrail"
+
+        client_info = request_data["metadata"].get(
+            "standard_logging_guardrail_information"
+        )
+        assert isinstance(client_info, list)
+        assert len(client_info) == 1
+
+    def test_no_duplicate_entry_when_metadata_aliases_litellm_metadata(self):
+        """If ``metadata`` and ``litellm_metadata`` reference the same dict, the
+        entry must be recorded once, not twice.
+        """
+        shared: dict = {}
+        request_data = {"metadata": shared, "litellm_metadata": shared}
+
+        self._invoke_add_log(request_data)
+
+        info = shared["standard_logging_guardrail_information"]
+        assert isinstance(info, list)
+        assert len(info) == 1
+
 
 class TestGuardrailOtelSpanEmission:
     """Recording a guardrail emits its otel span inline, so every guardrail