From 9d1bba72b92ac922a453cd206075cde9219508b4 Mon Sep 17 00:00:00 2001
From: yucheng <yucheng@berri.ai>
Date: Mon, 5 Oct 2026 10:23:00 +0000
Subject: [PATCH] fix(guardrails): keep the response scan when the request copy
 fails for explicit both scope

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 litellm/integrations/custom_guardrail.py      | 23 +++++++++++--------
 .../integrations/test_custom_guardrail.py     | 16 +++++++++++++
 2 files changed, 29 insertions(+), 10 deletions(-)

diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py
index 27e941dc245..94f1207b9ae 100644
--- a/litellm/integrations/custom_guardrail.py
+++ b/litellm/integrations/custom_guardrail.py
@@ -1008,16 +1008,23 @@ class CustomGuardrail(CustomLogger):
     def _copy_scratch_request_fields(
         self,
         kwargs: Mapping[str, object],
-    ) -> tuple[object, object]:
+    ) -> tuple[object, object] | None:
         optional_params: Final = kwargs.get("optional_params")
         try:
             return (
                 copy.deepcopy(kwargs.get("messages") or kwargs.get("input")),
                 copy.deepcopy(optional_params.get("tools") if isinstance(optional_params, Mapping) else None),
             )
-        except Exception:
+        except Exception as e:
             if self.logging_only_scope == "output":
-                return None, None
+                return None
+            if self.logging_only_scope == "both":
+                verbose_logger.warning(
+                    "Guardrail %s: logging_only request copy failed, skipping request scan: %s",
+                    self.guardrail_name,
+                    e,
+                )
+                return None
             raise
 
     async def _scan_logged_call(
@@ -1028,7 +1035,8 @@ class CustomGuardrail(CustomLogger):
         output_translation: "BaseTranslation",
         scratch_metadata: dict,  # mutable-ok: apply_guardrail records its verdict into request metadata
     ) -> None:
-        scratch_input, scratch_tools = self._copy_scratch_request_fields(kwargs)
+        scratch_fields: Final = self._copy_scratch_request_fields(kwargs)
+        scratch_input, scratch_tools = scratch_fields or (None, None)
         scratch_request: Final = {
             "model": kwargs.get("model"),
             "messages": scratch_input,
@@ -1037,13 +1045,8 @@ class CustomGuardrail(CustomLogger):
             "litellm_call_id": kwargs.get("litellm_call_id"),
             "metadata": scratch_metadata,
         }
-        if self.logging_only_scope != "output":
+        if self.logging_only_scope != "output" and scratch_fields is not None:
             if self.logging_only_scope == "both":
-                # An explicitly configured "both" observer asked for a verdict on
-                # each direction, so a failed request scan must not silently drop
-                # the response verdict. The implicit default (logging_only_scope
-                # None) keeps the abort semantics of a logging_only hook whose
-                # scan raised.
                 try:
                     await translation.process_input_messages(data=scratch_request, guardrail_to_apply=self)
                 except Exception as e:  # noqa: BLE001  # one direction's scan failure must not drop the other direction's verdict
diff --git a/tests/unit/integrations/test_custom_guardrail.py b/tests/unit/integrations/test_custom_guardrail.py
index fa47fb79820..72c362425bb 100644
--- a/tests/unit/integrations/test_custom_guardrail.py
+++ b/tests/unit/integrations/test_custom_guardrail.py
@@ -2647,6 +2647,22 @@ class TestLoggingOnlyApplyGuardrail:
         assert [entry["guardrail_name"] for entry in entries] == ["apply-only-observer"]
         assert [entry["guardrail_status"] for entry in entries] == ["success"]
 
+    @pytest.mark.asyncio
+    async def test_request_copy_failure_does_not_drop_the_response_scan_for_explicit_both_scope(self):
+        import threading
+
+        guardrail: Final = _ApplyOnlyObserver()
+        guardrail.logging_only_scope = "both"
+        call: Final = _logged_call([{"role": "user", "content": "hello there", "lock": threading.Lock()}])
+        kwargs: Final = call[0]
+        response: Final = call[1]
+
+        out_kwargs, _ = await guardrail.async_logging_hook(kwargs, response, CallTypes.acompletion.value)
+
+        assert guardrail.calls == [("response", ["general kenobi"])]
+        entries: Final = out_kwargs["standard_logging_object"]["guardrail_information"]
+        assert [entry["guardrail_status"] for entry in entries] == ["success"]
+
     @pytest.mark.asyncio
     async def test_block_verdict_is_recorded_without_raising(self):
         guardrail = _ApplyOnlyObserver(block=True)