refactor(grayswan): tighten post-call context typing and wire test helpers

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

litellm/proxy/guardrails/guardrail_hooks/grayswan/grayswan.py

@@ -617,7 +617,7 @@ class GraySwanGuardrail(CustomGuardrail):
         """
         from litellm.llms import load_guardrail_translation_mappings
 
-        call_type: Final = getattr(logging_obj, "call_type", None) or getattr(
+        call_type: Final = (logging_obj.call_type if logging_obj is not None else None) or getattr(
             request_data.get("litellm_logging_obj"), "call_type", None
         )
         if not isinstance(call_type, str):

tests/integration/observability/test_grayswan_wire.py

@@ -119,16 +119,17 @@ def _chat_provider(message: dict[str, JsonValue]):
 
 
 def _monitor_bodies(vendor: Wire, expected: int = 1) -> tuple[dict[str, JsonValue], ...]:
-    scans: Final = eventually(
-        lambda: tuple(
+    collected: list[dict[str, JsonValue]] = []  # mutable-ok: accumulator across polling attempts
+
+    def drain_new() -> tuple[dict[str, JsonValue], ...]:
+        collected.extend(
             _JSON_OBJECT.validate_json(request.body)
             for request in vendor.drain()
             if request.target == "/cygnal/monitor"
-        ),
-        lambda bodies: len(bodies) >= expected,
-        seconds=30,
-    )
-    return scans
+        )
+        return tuple(collected)
+
+    return eventually(drain_new, lambda bodies: len(bodies) >= expected, seconds=30)
 
 
 def test_post_call_sends_request_conversation_and_tools(gateway: Gateway, tmp_path: Path) -> None:

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

@@ -1,4 +1,3 @@
-
 import pytest
 from fastapi import HTTPException