Add pipeline validation in policy validator

litellm/proxy/policy_engine/policy_validator.py

@@ -283,8 +283,14 @@ class PolicyValidator:
                         )
                     )
 
-            # Note: Team, key, and model validation is done via policy_attachments
-            # Policies no longer have scope - attachments define where policies apply
+            # Validate pipeline if present
+            if policy.pipeline is not None:
+                pipeline_errors = PolicyValidator._validate_pipeline(
+                    policy_name=policy_name,
+                    policy=policy,
+                    available_guardrails=available_guardrails,
+                )
+                errors.extend(pipeline_errors)
 
             # Validate inheritance
             inheritance_errors = self._validate_inheritance_chain(
@@ -298,6 +304,53 @@ class PolicyValidator:
             warnings=warnings,
         )
 
+    @staticmethod
+    def _validate_pipeline(
+        policy_name: str,
+        policy: Policy,
+        available_guardrails: Set[str],
+    ) -> List[PolicyValidationError]:
+        """Validate a policy's pipeline configuration."""
+        errors: List[PolicyValidationError] = []
+        pipeline = policy.pipeline
+        if pipeline is None:
+            return errors
+
+        guardrails_add = set(policy.guardrails.get_add())
+
+        for i, step in enumerate(pipeline.steps):
+            # Check guardrail is in policy's guardrails.add
+            if step.guardrail not in guardrails_add:
+                errors.append(
+                    PolicyValidationError(
+                        policy_name=policy_name,
+                        error_type=PolicyValidationErrorType.INVALID_GUARDRAIL,
+                        message=(
+                            f"Pipeline step {i} guardrail '{step.guardrail}' "
+                            f"is not in the policy's guardrails.add list"
+                        ),
+                        field="pipeline.steps",
+                        value=step.guardrail,
+                    )
+                )
+
+            # Check guardrail exists in registry
+            if available_guardrails and step.guardrail not in available_guardrails:
+                errors.append(
+                    PolicyValidationError(
+                        policy_name=policy_name,
+                        error_type=PolicyValidationErrorType.INVALID_GUARDRAIL,
+                        message=(
+                            f"Pipeline step {i} guardrail '{step.guardrail}' "
+                            f"not found in guardrail registry"
+                        ),
+                        field="pipeline.steps",
+                        value=step.guardrail,
+                    )
+                )
+
+        return errors
+
     async def validate_policy_config(
         self,
         policy_config: Dict[str, Any],