fix(guardrails): keep output-only scans when request copy fails

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

litellm/integrations/custom_guardrail.py

@@ -990,6 +990,21 @@ class CustomGuardrail(CustomLogger):
             "standard_logging_object": {**standard_logging_object, "guardrail_information": [*existing, *entries]},
         }, result
 
+    def _copy_scratch_request_fields(
+        self,
+        kwargs: dict,  # mutable-ok: CustomLogger.async_logging_hook contract
+    ) -> tuple[object, object]:
+        optional_params: Final = kwargs.get("optional_params") or {}
+        try:
+            return (
+                copy.deepcopy(kwargs.get("messages") or kwargs.get("input")),
+                copy.deepcopy(optional_params.get("tools")),
+            )
+        except Exception:
+            if self.logging_only_scope == "output":
+                return None, None
+            raise
+
     async def _scan_logged_call(
         self,
         kwargs: dict,  # mutable-ok: CustomLogger.async_logging_hook contract
@@ -998,13 +1013,12 @@ class CustomGuardrail(CustomLogger):
         output_translation: "BaseTranslation",
         scratch_metadata: dict,  # mutable-ok: apply_guardrail records its verdict into request metadata
     ) -> None:
-        optional_params: Final = kwargs.get("optional_params") or {}
-        scratch_input: Final = copy.deepcopy(kwargs.get("messages") or kwargs.get("input"))
+        scratch_input, scratch_tools = self._copy_scratch_request_fields(kwargs)
         scratch_request: Final = {
             "model": kwargs.get("model"),
             "messages": scratch_input,
             "input": scratch_input,
-            "tools": copy.deepcopy(optional_params.get("tools")),
+            "tools": scratch_tools,
             "litellm_call_id": kwargs.get("litellm_call_id"),
             "metadata": scratch_metadata,
         }

tests/unit/integrations/test_custom_guardrail.py

@@ -2630,6 +2630,23 @@ class TestLoggingOnlyApplyGuardrail:
         assert out_kwargs is kwargs
         assert out_response is response
 
+    @pytest.mark.asyncio
+    async def test_output_scope_scans_response_when_request_copy_fails(self):
+        import threading
+
+        guardrail: Final = _ApplyOnlyObserver()
+        guardrail.logging_only_scope = "output"
+        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_name"] for entry in entries] == ["apply-only-observer"]
+        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)