chore(guardrails): align Reco with main's CI gates

Upstream moved unit tests from tests/test_litellm to tests/unit, so
test_reco.py was invoked by no CI job. Move it next to the other
guardrail tests

Replace RecoGuardrail's forwarded kwargs with explicit keyword
params, read optional_params through a pydantic adapter instead of
untyped dict/getattr access, freeze RecoOptionalParams, and drop
mutable-ok markers that suppress nothing, to stay within main's
type-discipline and basedpyright budgets

Signed-off-by: MickeyShnaiderman-RecoLabs <mickeys@reco.ai>
Co-authored-by: Cursor <cursoragent@cursor.com>

litellm/proxy/guardrails/guardrail_hooks/reco/__init__.py

@@ -1,5 +1,9 @@
+from collections.abc import Mapping
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Final
 
+from pydantic import BaseModel, TypeAdapter
+
 from litellm.secret_managers.main import get_secret_str
 from litellm.types.guardrails import SupportedGuardrailIntegrations
 from litellm.types.proxy.guardrails.guardrail_hooks.reco import RecoOptionalParams
@@ -9,18 +13,19 @@ from .reco import RecoGuardrail
 if TYPE_CHECKING:
     from litellm.types.guardrails import Guardrail, LitellmParams
 
+_CONFIG_MAPPING: Final = TypeAdapter(Mapping[str, object])
 
-def _get_config_value(litellm_params: "LitellmParams", optional_params: object, attribute_name: str) -> object:
-    """Read one param from optional_params (a dict or a pydantic object), falling back to a top-level
-    litellm_params field only when it was set explicitly, since those carry fail-closed defaults.
-    Resolves an ``os.environ/<VAR>`` reference the way guardrail api_key/api_base are resolved."""
-    nested: Final = (
-        optional_params.get(attribute_name)
-        if isinstance(optional_params, dict)
-        else getattr(optional_params, attribute_name, None)
-    )
-    explicitly_set_top_level: Final = attribute_name in getattr(litellm_params, "model_fields_set", frozenset())
-    value: Final = getattr(litellm_params, attribute_name) if nested is None and explicitly_set_top_level else nested
+
+def _optional_params_mapping(optional_params: object) -> Mapping[str, object]:
+    """optional_params arrives as a pydantic object from YAML, or as a plain dict when reconciled from the DB."""
+    if optional_params is None:
+        return MappingProxyType({})
+    raw: Final = optional_params.model_dump() if isinstance(optional_params, BaseModel) else optional_params
+    return _CONFIG_MAPPING.validate_python(raw)
+
+
+def _first_configured(field_name: str, sources: tuple[Mapping[str, object], ...]) -> object:
+    value: Final = next((source[field_name] for source in sources if source.get(field_name) is not None), None)
     if isinstance(value, str) and value.startswith("os.environ/"):
         return get_secret_str(value)
     return value
@@ -29,34 +34,37 @@ def _get_config_value(litellm_params: "LitellmParams", optional_params: object,
 def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail") -> RecoGuardrail:
     import litellm
 
-    optional_params: Final = getattr(litellm_params, "optional_params", None)
+    # Top-level fields only count when set explicitly: their defaults are fail-closed and would flip Reco's.
+    explicit_top_level: Final = _CONFIG_MAPPING.validate_python(
+        litellm_params.model_dump(include=litellm_params.model_fields_set & RecoOptionalParams.model_fields.keys())
+    )
+    sources: Final = (_optional_params_mapping(litellm_params.optional_params), explicit_top_level)
     configured: Final = (
-        (field_name, _get_config_value(litellm_params, optional_params, field_name))
-        for field_name in RecoOptionalParams.model_fields
+        (field_name, _first_configured(field_name, sources)) for field_name in RecoOptionalParams.model_fields
     )
     reco_params: Final = RecoOptionalParams.model_validate(
         {field_name: value for field_name, value in configured if value is not None}
     )
 
     _reco_callback: Final = RecoGuardrail(
-        guardrail_name=guardrail.get("guardrail_name", ""),
+        guardrail_name=guardrail["guardrail_name"],
         reco_tenant_id=reco_params.reco_tenant_id,
         api_base=reco_params.api_base,
         unreachable_fallback=reco_params.unreachable_fallback,
         fail_on_error=reco_params.fail_on_error,
         event_hook=litellm_params.mode,
-        default_on=litellm_params.default_on,
+        default_on=bool(litellm_params.default_on),
     )
     litellm.logging_callback_manager.add_litellm_callback(_reco_callback)
 
     return _reco_callback
 
 
-guardrail_initializer_registry: Final = {  # mutable-ok: guardrail auto-discovery requires an actual dict instance
+guardrail_initializer_registry: Final = {
     SupportedGuardrailIntegrations.RECO.value: initialize_guardrail,
 }
 
 
-guardrail_class_registry: Final = {  # mutable-ok: guardrail auto-discovery requires an actual dict instance
+guardrail_class_registry: Final = {
     SupportedGuardrailIntegrations.RECO.value: RecoGuardrail,
 }

litellm/proxy/guardrails/guardrail_hooks/reco/reco.py

@@ -1,27 +1,21 @@
-from collections.abc import Mapping
-from typing import TYPE_CHECKING, Literal, Optional
-
-from typing_extensions import ReadOnly, TypedDict, Unpack
+from collections.abc import Mapping, Sequence
+from typing import TYPE_CHECKING, Final, Literal, Optional
 
 from litellm.exceptions import GuardrailRaisedException
-from litellm.integrations.custom_guardrail import log_guardrail_information
+from litellm.integrations.custom_guardrail import (
+    log_guardrail_information,  # pyright: ignore[reportUnknownVariableType]  # decorator is untyped
+)
 from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.generic_guardrail_api import (
     GenericGuardrailAPI,
 )
-from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.guardrails import GuardrailEventHooks, Mode
 from litellm.types.proxy.guardrails.guardrail_hooks.reco import validate_reco_tenant_id
 from litellm.types.utils import GenericGuardrailAPIInputs
 
 if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
     from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
-
-
-class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
-    """Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
-
-    guardrail_name: ReadOnly[str | None]
-    default_on: ReadOnly[bool]
+    from litellm.types.proxy.guardrails.guardrail_hooks.reco import RecoOptionalParams
 
 
 class RecoGuardrail(GenericGuardrailAPI):
@@ -34,30 +28,32 @@ class RecoGuardrail(GenericGuardrailAPI):
         headers: Mapping[str, str] | None = None,
         unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_open",
         fail_on_error: bool = False,
-        **kwargs: Unpack[_CustomGuardrailOptions],
+        guardrail_name: str | None = None,
+        event_hook: str | Sequence[str] | Mode | None = None,
+        default_on: bool = False,
     ) -> None:
         if not reco_tenant_id:
             raise ValueError("reco_tenant_id is required for the Reco guardrail")
         if not api_base:
             raise ValueError("api_base is required for the Reco guardrail")
 
-        validated_tenant_id = validate_reco_tenant_id(reco_tenant_id)
-        base_headers = headers or {}  # mutable-ok: normalizes headers to a dict
-        merged_headers = {**base_headers, "X-Reco-Tenant-Id": validated_tenant_id}  # mutable-ok: required by base class
+        validated_tenant_id: Final = validate_reco_tenant_id(reco_tenant_id)
 
-        super().__init__(
-            headers=merged_headers,
+        super().__init__(  # pyright: ignore[reportUnknownMemberType]  # parent takes untyped extra_headers and **kwargs
+            headers={**(headers or {}), "X-Reco-Tenant-Id": validated_tenant_id},
             api_base=api_base,
             unreachable_fallback=unreachable_fallback,
             fail_on_error=fail_on_error,
-            **kwargs,
+            guardrail_name=guardrail_name,
+            event_hook=event_hook,
+            default_on=default_on,
         )
 
     @log_guardrail_information
     async def apply_guardrail(
         self,
         inputs: GenericGuardrailAPIInputs,
-        request_data: dict,  # mutable-ok: must match GenericGuardrailAPI.apply_guardrail's own param type
+        request_data: dict[str, object],  # mutable-ok: must match GenericGuardrailAPI.apply_guardrail's own param type
         input_type: Literal["request", "response"],
         logging_obj: Optional["LiteLLMLoggingObj"] = None,
     ) -> GenericGuardrailAPIInputs:
@@ -65,7 +61,9 @@ class RecoGuardrail(GenericGuardrailAPI):
         # not the guardrail's own configured name. Re-raise with the right one instead of
         # patching the shared base class.
         try:
-            return await super().apply_guardrail(inputs, request_data, input_type, logging_obj)
+            return await super().apply_guardrail(  # pyright: ignore[reportUnknownMemberType]  # parent annotates request_data as a bare dict
+                inputs, request_data, input_type, logging_obj
+            )
         except GuardrailRaisedException as e:
             raise GuardrailRaisedException(
                 guardrail_name=self.guardrail_name,
@@ -76,11 +74,11 @@ class RecoGuardrail(GenericGuardrailAPI):
             ) from e
 
     @staticmethod
-    def get_config_model() -> type["GuardrailConfigModel"] | None:
+    def get_config_model() -> type["GuardrailConfigModel[RecoOptionalParams]"] | None:
         from litellm.types.proxy.guardrails.guardrail_hooks.reco import RecoConfigModel
 
         return RecoConfigModel
 
     @classmethod
     def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:  # mutable-ok: fixed by base class
-        return [GuardrailEventHooks.pre_call]  # mutable-ok: fixed by base class
+        return [GuardrailEventHooks.pre_call]

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

@@ -1,7 +1,7 @@
 import uuid
 from typing import Literal
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from .base import GuardrailConfigModel
 
@@ -17,6 +17,8 @@ def validate_reco_tenant_id(value: str) -> str:
 class RecoOptionalParams(BaseModel):
     """Configuration parameters for the Reco guardrail"""
 
+    model_config = ConfigDict(frozen=True)
+
     reco_tenant_id: str = Field(
         description="Tenant identifier for the Reco account, as a UUID. Sent as the X-Reco-Tenant-Id header on every guardrail request.",
     )

tests/unit/proxy/guardrails/guardrail_hooks/test_reco.py

@@ -2,7 +2,6 @@
 Tests for the Reco guardrail integration.
 """
 
-from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import httpx
@@ -95,7 +94,8 @@ class TestRecoGuardrailInitializer:
         """Guardrails reconciled from the DB can carry `optional_params` as a
         plain dict rather than a pydantic object; attribute-only access would
         silently read None for every field."""
-        litellm_params = SimpleNamespace(
+        litellm_params = LitellmParams.model_construct(
+            guardrail="reco",
             optional_params={
                 "reco_tenant_id": "33333333-3333-3333-3333-333333333333",
                 "api_base": "https://edge3.eu.reco.ai",
@@ -111,7 +111,8 @@ class TestRecoGuardrailInitializer:
 
     def test_initialize_guardrail_resolves_env_var_reference(self, monkeypatch):
         monkeypatch.setenv("RECO_TEST_TENANT_ID", "44444444-4444-4444-4444-444444444444")
-        litellm_params = SimpleNamespace(
+        litellm_params = LitellmParams(
+            guardrail="reco",
             optional_params={
                 "reco_tenant_id": "os.environ/RECO_TEST_TENANT_ID",
                 "api_base": "https://edge4.eu.reco.ai",