diff --git a/packages/uipath-platform/pyproject.toml b/packages/uipath-platform/pyproject.toml index c5cdf13ee..ba5634ef1 100644 --- a/packages/uipath-platform/pyproject.toml +++ b/packages/uipath-platform/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "uipath-platform" -version = "0.1.17" +version = "0.1.18" description = "HTTP client library for programmatic access to UiPath Platform" readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.11" diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/__init__.py b/packages/uipath-platform/src/uipath/platform/guardrails/__init__.py index ffab74581..c389812d1 100644 --- a/packages/uipath-platform/src/uipath/platform/guardrails/__init__.py +++ b/packages/uipath-platform/src/uipath/platform/guardrails/__init__.py @@ -14,6 +14,24 @@ ) from ._guardrails_service import GuardrailsService +from .decorators import ( + BlockAction, + CustomValidator, + GuardrailAction, + GuardrailBlockException, + GuardrailExecutionStage, + GuardrailTargetAdapter, + GuardrailValidatorBase, + LogAction, + LoggingSeverityLevel, + PIIDetectionEntity, + PIIDetectionEntityType, + PIIValidator, + PromptInjectionValidator, + RuleFunction, + guardrail, + register_guardrail_adapter, +) from .guardrails import ( BuiltInValidatorGuardrail, EnumListParameterValue, @@ -22,7 +40,9 @@ ) __all__ = [ + # Service "GuardrailsService", + # Guardrail models "BuiltInValidatorGuardrail", "GuardrailType", "GuardrailValidationResultType", @@ -33,4 +53,21 @@ "GuardrailValidationResult", "EnumListParameterValue", "MapEnumParameterValue", + # Decorator framework + "guardrail", + "GuardrailValidatorBase", + "PIIValidator", + "PromptInjectionValidator", + "CustomValidator", + "RuleFunction", + "PIIDetectionEntity", + "PIIDetectionEntityType", + "GuardrailExecutionStage", + "GuardrailAction", + "LogAction", + "BlockAction", + "LoggingSeverityLevel", + "GuardrailBlockException", + "GuardrailTargetAdapter", + "register_guardrail_adapter", ] diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/__init__.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/__init__.py new file mode 100644 index 000000000..68f188eaf --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/__init__.py @@ -0,0 +1,64 @@ +"""Guardrail decorator framework for UiPath Platform. + +Provides the ``@guardrail`` decorator, built-in validators, actions, and an +adapter registry that framework integrations (e.g. *uipath-langchain*) use to +teach the decorator how to wrap their specific object types. + +Quick start:: + + from uipath.platform.guardrails.decorators import ( + guardrail, + PIIValidator, + LogAction, + BlockAction, + PIIDetectionEntity, + PIIDetectionEntityType, + GuardrailExecutionStage, + ) + + pii = PIIValidator(entities=[PIIDetectionEntity(PIIDetectionEntityType.EMAIL)]) + + # Applied to a factory function (LangChain adapter must be imported first): + @guardrail(validator=pii, action=LogAction(), name="LLM PII") + def create_llm(): + ... +""" + +from ._actions import BlockAction, LogAction, LoggingSeverityLevel +from ._enums import GuardrailExecutionStage, PIIDetectionEntityType +from ._exceptions import GuardrailBlockException +from ._guardrail import guardrail +from ._models import GuardrailAction, PIIDetectionEntity +from ._registry import GuardrailTargetAdapter, register_guardrail_adapter +from .validators import ( + CustomValidator, + GuardrailValidatorBase, + PIIValidator, + PromptInjectionValidator, + RuleFunction, +) + +__all__ = [ + # Decorator + "guardrail", + # Validators + "GuardrailValidatorBase", + "PIIValidator", + "PromptInjectionValidator", + "CustomValidator", + "RuleFunction", + # Models & enums + "PIIDetectionEntity", + "PIIDetectionEntityType", + "GuardrailExecutionStage", + "GuardrailAction", + # Actions + "LogAction", + "BlockAction", + "LoggingSeverityLevel", + # Exception + "GuardrailBlockException", + # Adapter registry + "GuardrailTargetAdapter", + "register_guardrail_adapter", +] diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_actions.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_actions.py new file mode 100644 index 000000000..8e6489797 --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_actions.py @@ -0,0 +1,82 @@ +"""Built-in GuardrailAction implementations.""" + +import logging +from dataclasses import dataclass +from enum import Enum +from typing import Any, Optional + +from uipath.core.guardrails import ( + GuardrailValidationResult, + GuardrailValidationResultType, +) + +from ._exceptions import GuardrailBlockException +from ._models import GuardrailAction + + +class LoggingSeverityLevel(int, Enum): + """Logging severity level for :class:`LogAction`.""" + + ERROR = logging.ERROR + INFO = logging.INFO + WARNING = logging.WARNING + DEBUG = logging.DEBUG + + +@dataclass +class LogAction(GuardrailAction): + """Log guardrail violations without stopping execution. + + Args: + severity_level: Python logging level. Defaults to ``WARNING``. + message: Custom log message. If omitted, the validation reason is used. + """ + + severity_level: LoggingSeverityLevel = LoggingSeverityLevel.WARNING + message: Optional[str] = None + + def handle_validation_result( + self, + result: GuardrailValidationResult, + data: str | dict[str, Any], + guardrail_name: str, + ) -> str | dict[str, Any] | None: + """Log the violation and return ``None`` (no data modification).""" + if result.result == GuardrailValidationResultType.VALIDATION_FAILED: + msg = self.message or f"Failed: {result.reason}" + logging.getLogger(__name__).log( + self.severity_level, + "[GUARDRAIL] [%s] %s", + guardrail_name, + msg, + ) + return None + + +@dataclass +class BlockAction(GuardrailAction): + """Block execution by raising :class:`GuardrailBlockException`. + + Framework adapters catch ``GuardrailBlockException`` at the wrapper boundary + and convert it to their own runtime error type. + + Args: + title: Exception title. Defaults to a message derived from the guardrail name. + detail: Exception detail. Defaults to the validation reason. + """ + + title: Optional[str] = None + detail: Optional[str] = None + + def handle_validation_result( + self, + result: GuardrailValidationResult, + data: str | dict[str, Any], + guardrail_name: str, + ) -> str | dict[str, Any] | None: + """Raise :class:`GuardrailBlockException` when validation fails.""" + if result.result == GuardrailValidationResultType.VALIDATION_FAILED: + title = self.title or f"Guardrail [{guardrail_name}] blocked execution" + detail = self.detail or result.reason or "Guardrail validation failed" + raise GuardrailBlockException(title=title, detail=detail) + return None diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_core.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_core.py new file mode 100644 index 000000000..b843b416a --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_core.py @@ -0,0 +1,242 @@ +"""Core framework-agnostic utilities for guardrail decorators.""" + +import ast +import inspect +import json +import logging +from typing import Any, Callable + +from uipath.core.guardrails import ( + GuardrailValidationResult, + GuardrailValidationResultType, +) + +from uipath.platform.guardrails.guardrails import BuiltInValidatorGuardrail + +from ._enums import GuardrailExecutionStage + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Evaluator type alias +# --------------------------------------------------------------------------- + +_EvaluatorFn = Callable[ + [ + "str | dict[str, Any]", # data + GuardrailExecutionStage, # stage + "dict[str, Any] | None", # input_data + "dict[str, Any] | None", # output_data + ], + GuardrailValidationResult, +] +"""Type alias for the unified evaluation callable used by all wrappers.""" + + +# --------------------------------------------------------------------------- +# Evaluator factory +# --------------------------------------------------------------------------- + + +def _make_evaluator( + validator: Any, + built_in_guardrail: BuiltInValidatorGuardrail | None, +) -> _EvaluatorFn: + """Return a unified evaluation callable. + + If *built_in_guardrail* is provided the callable hits the UiPath API (lazily + initializing ``UiPath()``). Otherwise, it delegates to ``validator.evaluate()``. + + Args: + validator: :class:`GuardrailValidatorBase` instance for local evaluation. + built_in_guardrail: Pre-built ``BuiltInValidatorGuardrail``, or ``None``. + + Returns: + Callable with signature ``(data, stage, input_data, output_data)``. + """ + if built_in_guardrail is not None: + _uipath_holder: list[Any] = [] + + def _api_eval( + data: str | dict[str, Any], + stage: GuardrailExecutionStage, + input_data: dict[str, Any] | None, + output_data: dict[str, Any] | None, + ) -> GuardrailValidationResult: + if not _uipath_holder: + from uipath.platform import UiPath + + _uipath_holder.append(UiPath()) + return _uipath_holder[0].guardrails.evaluate_guardrail( + data, built_in_guardrail + ) + + return _api_eval + + def _local_eval( + data: str | dict[str, Any], + stage: GuardrailExecutionStage, + input_data: dict[str, Any] | None, + output_data: dict[str, Any] | None, + ) -> GuardrailValidationResult: + return validator.evaluate(data, stage, input_data, output_data) + + return _local_eval + + +# --------------------------------------------------------------------------- +# Guardrail evaluation helpers +# --------------------------------------------------------------------------- + + +# --------------------------------------------------------------------------- +# Tool I/O normalisation helpers +# --------------------------------------------------------------------------- + + +def _is_tool_call_envelope(tool_input: Any) -> bool: + """Return ``True`` if *tool_input* is a LangGraph tool-call envelope dict.""" + return ( + isinstance(tool_input, dict) + and "args" in tool_input + and tool_input.get("type") == "tool_call" + ) + + +def _extract_input(tool_input: Any) -> dict[str, Any]: + """Normalise tool input to a plain dict for rule / guardrail evaluation. + + LangGraph wraps tool inputs as ``{"name": ..., "args": {...}, "type": "tool_call"}``. + This function unwraps ``args`` so rules can access the actual tool arguments. + """ + if _is_tool_call_envelope(tool_input): + args = tool_input["args"] + if isinstance(args, dict): + return args + if isinstance(tool_input, dict): + return tool_input + return {"input": tool_input} + + +def _rewrap_input(original_tool_input: Any, modified_args: dict[str, Any]) -> Any: + """Re-wrap modified args back into the original tool-call envelope (if applicable).""" + if _is_tool_call_envelope(original_tool_input): + import copy + + wrapped = copy.copy(original_tool_input) + wrapped["args"] = modified_args + return wrapped + return modified_args + + +def _extract_output(result: Any) -> dict[str, Any]: + """Normalise tool output to a dict for guardrail / rule evaluation. + + This is the framework-agnostic version. Framework adapters that deal with + framework-specific envelope types (e.g. LangGraph ``ToolMessage`` / + ``Command``) should pre-process the result before calling this function. + + Falls back to ``{"output": content}`` for plain strings and anything else. + """ + if isinstance(result, dict): + return result + if isinstance(result, str): + try: + parsed = json.loads(result) + return parsed if isinstance(parsed, dict) else {"output": parsed} + except ValueError: + try: + parsed = ast.literal_eval(result) + return parsed if isinstance(parsed, dict) else {"output": parsed} + except (ValueError, SyntaxError): + return {"output": result} + return {"output": result} + + +# --------------------------------------------------------------------------- +# Rule evaluation (used by CustomValidator) +# --------------------------------------------------------------------------- + + +def _evaluate_rules( + rules: "list[Callable[..., bool]]", + stage: GuardrailExecutionStage, + input_data: "dict[str, Any] | None", + output_data: "dict[str, Any] | None", + guardrail_name: str = "Rule", +) -> GuardrailValidationResult: + """Evaluate a list of deterministic rule functions and return a result. + + Semantics: + - Empty rules → always fail (trigger action). + - All rules must detect a violation to fail; any passing rule → PASSED. + - Rules with the wrong parameter count for the current stage are skipped. + + Args: + rules: List of callables returning ``True`` on violation. + stage: Current execution stage (PRE or POST). + input_data: Normalised tool input dict. + output_data: Normalised tool output dict (``None`` at PRE stage). + guardrail_name: Used in reason strings. + + Returns: + ``GuardrailValidationResult`` with PASSED or VALIDATION_FAILED. + """ + if not rules: + return GuardrailValidationResult( + result=GuardrailValidationResultType.VALIDATION_FAILED, + reason="Empty rules — always apply action", + ) + + violations: list[str] = [] + passed_rules: list[str] = [] + evaluated_count = 0 + + for rule in rules: + try: + sig = inspect.signature(rule) + param_count = len(sig.parameters) + + if stage == GuardrailExecutionStage.PRE: + if input_data is None or param_count != 1: + continue + violation = rule(input_data) + evaluated_count += 1 + else: + if output_data is None: + continue + if param_count == 2 and input_data is not None: + violation = rule(input_data, output_data) + elif param_count == 1: + violation = rule(output_data) + else: + continue + evaluated_count += 1 + + if violation: + violations.append(f"Rule {guardrail_name} detected violation") + else: + passed_rules.append(f"Rule {guardrail_name}") + except Exception as exc: + logger.error( + "Error in rule function %s: %s", guardrail_name, exc, exc_info=True + ) + violations.append(f"Rule {guardrail_name} raised exception: {exc!s}") + evaluated_count += 1 + + if evaluated_count == 0: + return GuardrailValidationResult( + result=GuardrailValidationResultType.PASSED, + reason="No applicable rules to evaluate", + ) + + if passed_rules: + return GuardrailValidationResult( + result=GuardrailValidationResultType.PASSED, + reason=f"Rules passed: {', '.join(passed_rules)}", + ) + + return GuardrailValidationResult( + result=GuardrailValidationResultType.VALIDATION_FAILED, + reason="; ".join(violations), + ) diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_enums.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_enums.py new file mode 100644 index 000000000..be7832ddf --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_enums.py @@ -0,0 +1,44 @@ +"""Enums for guardrail decorators.""" + +from enum import Enum + + +class GuardrailExecutionStage(str, Enum): + """Execution stage for guardrails.""" + + PRE = "pre" + """Evaluate before the target executes.""" + + POST = "post" + """Evaluate after the target executes.""" + + PRE_AND_POST = "pre&post" + """Evaluate both before and after the target executes.""" + + +class PIIDetectionEntityType(str, Enum): + """PII detection entity types supported by UiPath guardrails. + + These entities match the available options from the UiPath guardrails service + backend. The enum values correspond to the exact strings expected by the API. + """ + + PERSON = "Person" + ADDRESS = "Address" + DATE = "Date" + PHONE_NUMBER = "PhoneNumber" + EUGPS_COORDINATES = "EugpsCoordinates" + EMAIL = "Email" + CREDIT_CARD_NUMBER = "CreditCardNumber" + INTERNATIONAL_BANKING_ACCOUNT_NUMBER = "InternationalBankingAccountNumber" + SWIFT_CODE = "SwiftCode" + ABA_ROUTING_NUMBER = "ABARoutingNumber" + US_DRIVERS_LICENSE_NUMBER = "USDriversLicenseNumber" + UK_DRIVERS_LICENSE_NUMBER = "UKDriversLicenseNumber" + US_INDIVIDUAL_TAXPAYER_IDENTIFICATION = "USIndividualTaxpayerIdentification" + UK_UNIQUE_TAXPAYER_NUMBER = "UKUniqueTaxpayerNumber" + US_BANK_ACCOUNT_NUMBER = "USBankAccountNumber" + US_SOCIAL_SECURITY_NUMBER = "USSocialSecurityNumber" + USUK_PASSPORT_NUMBER = "UsukPassportNumber" + URL = "URL" + IP_ADDRESS = "IPAddress" diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_exceptions.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_exceptions.py new file mode 100644 index 000000000..f4b7672e5 --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_exceptions.py @@ -0,0 +1,18 @@ +"""Exceptions for guardrail decorators.""" + + +class GuardrailBlockException(Exception): + """Raised by BlockAction when a guardrail blocks execution. + + Framework adapters (e.g. LangChain) should catch this and convert it to + their own runtime exception type at the outermost wrapper boundary. + + Args: + title: Brief title for the block event. + detail: Detailed reason for the block. + """ + + def __init__(self, title: str, detail: str) -> None: + self.title = title + self.detail = detail + super().__init__(f"{title}: {detail}") diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_guardrail.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_guardrail.py new file mode 100644 index 000000000..cd0e1c34f --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_guardrail.py @@ -0,0 +1,163 @@ +"""Single ``@guardrail`` decorator for all guardrail types.""" + +import inspect +from functools import wraps +from typing import Any + +from uipath.core.guardrails import GuardrailScope + +from ._core import _EvaluatorFn, _make_evaluator +from ._enums import GuardrailExecutionStage +from ._models import GuardrailAction +from ._registry import detect_scope_from_adapters, wrap_with_adapter +from .validators._base import GuardrailValidatorBase + +# --------------------------------------------------------------------------- +# Factory function wrapper +# --------------------------------------------------------------------------- + + +def _wrap_factory_callable( + func: Any, + evaluator: _EvaluatorFn, + action: GuardrailAction, + name: str, + stage: GuardrailExecutionStage, +) -> Any: + """Wrap a factory function so its return value is guarded via registered adapters. + + After calling *func*, the return value's type is inspected. If a registered + adapter recognises it, that adapter wraps it. Otherwise the return value is + passed through unchanged. + + Args: + func: Factory callable to wrap. + evaluator: Unified evaluation callable. + action: Action to invoke on violation. + name: Guardrail name. + stage: Execution stage. + + Returns: + Wrapped callable (sync or async, matching the original). + """ + + def _dispatch(result: Any) -> Any: + return wrap_with_adapter(result, evaluator, action, name, stage) + + if inspect.iscoroutinefunction(func): + + @wraps(func) + async def _wrapped_async(*args: Any, **kwargs: Any) -> Any: + return _dispatch(await func(*args, **kwargs)) + + return _wrapped_async + + @wraps(func) + def _wrapped(*args: Any, **kwargs: Any) -> Any: + return _dispatch(func(*args, **kwargs)) + + return _wrapped + + +# --------------------------------------------------------------------------- +# Public @guardrail decorator +# --------------------------------------------------------------------------- + + +def guardrail( + func: Any = None, + *, + validator: GuardrailValidatorBase, + action: GuardrailAction, + name: str = "Guardrail", + description: str | None = None, + stage: GuardrailExecutionStage = GuardrailExecutionStage.PRE_AND_POST, + enabled_for_evals: bool = True, +) -> Any: + """Apply a guardrail to a tool, LLM, or agent factory function. + + Scope is auto-detected from the decorated object via registered framework + adapters and validated against the validator's supported scopes and stages + at decoration time. Multiple ``@guardrail`` decorators can be stacked on + the same object. + + Args: + func: Object to decorate. Supplied directly when used without parentheses. + validator: :class:`~.validators.GuardrailValidatorBase` defining what to check. + action: :class:`~._models.GuardrailAction` defining how to respond on violation. + name: Human-readable name for this guardrail instance. + description: Optional description passed to API-based guardrails. + stage: When to evaluate — ``PRE``, ``POST``, or ``PRE_AND_POST``. + Defaults to ``PRE_AND_POST``. + enabled_for_evals: Whether this guardrail is active in evaluation scenarios. + Defaults to ``True``. + + Returns: + The decorated object. + + Raises: + ValueError: If *action* is invalid, or the validator does not support + the detected scope or requested stage. + """ + if action is None: + raise ValueError("action must be provided") + if not isinstance(action, GuardrailAction): + raise ValueError("action must be an instance of GuardrailAction") + if not isinstance(enabled_for_evals, bool): + raise ValueError("enabled_for_evals must be a boolean") + + def _apply(obj: Any) -> Any: + # ------------------------------------------------------------------ + # Determine whether a registered adapter recognises this object. + # ------------------------------------------------------------------ + adapter_scope = detect_scope_from_adapters(obj) + + if adapter_scope is not None: + # Adapter-registered type (e.g. LangChain BaseTool, BaseChatModel, …) + validator.validate_scope(adapter_scope) + validator.validate_stage(stage) + built_in_guardrail = validator.get_built_in_guardrail( + adapter_scope, name, description, enabled_for_evals + ) + evaluator = _make_evaluator(validator, built_in_guardrail) + return wrap_with_adapter(obj, evaluator, action, name, stage) + + # ------------------------------------------------------------------ + # Plain callable — treat as a factory function. + # ------------------------------------------------------------------ + if callable(obj): + # TOOL-only validators cannot be applied to factory functions because + # the scope (and thus whether the returned object is a tool) is only + # known at call time. + if validator.supported_scopes and all( + s == GuardrailScope.TOOL for s in validator.supported_scopes + ): + raise ValueError( + f"@guardrail with {type(validator).__name__} can only be applied " + "to BaseTool instances. " + "Apply it directly to the tool, not to a factory function." + ) + validator.validate_stage(stage) + # Use the validator's primary scope (first supported, or AGENT as default) + # to build the API guardrail instance for factory functions. + api_scope = ( + validator.supported_scopes[0] + if validator.supported_scopes + else GuardrailScope.AGENT + ) + built_in_guardrail = validator.get_built_in_guardrail( + api_scope, name, description, enabled_for_evals + ) + evaluator = _make_evaluator(validator, built_in_guardrail) + return _wrap_factory_callable(obj, evaluator, action, name, stage) + + raise ValueError( + f"@guardrail cannot be applied to {type(obj)!r}. " + "Target must be a framework-registered type (e.g. a LangChain BaseTool, " + "BaseChatModel, StateGraph) or a callable factory function. " + "Ensure the relevant framework adapter is imported before using @guardrail." + ) + + if func is None: + return _apply + return _apply(func) diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_models.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_models.py new file mode 100644 index 000000000..ac22538e0 --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_models.py @@ -0,0 +1,59 @@ +"""Models for guardrail decorators.""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any + +from uipath.core.guardrails import GuardrailValidationResult + + +@dataclass +class PIIDetectionEntity: + """PII entity configuration with detection threshold. + + Args: + name: The entity type name (e.g. ``PIIDetectionEntityType.EMAIL``). + threshold: Confidence threshold (0.0 to 1.0) for detection. + """ + + name: str + threshold: float = 0.5 + + def __post_init__(self) -> None: + if not 0.0 <= self.threshold <= 1.0: + raise ValueError( + f"Threshold must be between 0.0 and 1.0, got {self.threshold}" + ) + + +class GuardrailAction(ABC): + """Interface for defining custom actions when a guardrail violation is detected. + + Subclass this to implement custom behaviour on validation failure, such as + logging, blocking, or content sanitisation. Built-in implementations are + :class:`LogAction` and :class:`BlockAction`. + """ + + @abstractmethod + def handle_validation_result( + self, + result: GuardrailValidationResult, + data: str | dict[str, Any], + guardrail_name: str, + ) -> "str | dict[str, Any] | None": + """Handle a guardrail validation result. + + Called when guardrail validation fails. May return modified data to + sanitise/filter the validated content before execution continues, or + ``None`` to leave it unchanged. + + Args: + result: The validation result from the guardrails service. + data: The data that was validated (string or dictionary). Depending + on context this can be tool input, tool output, or message text. + guardrail_name: The name of the guardrail that triggered. + + Returns: + Modified data if the action wants to replace the original, or + ``None`` if no modification is needed. + """ diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_registry.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_registry.py new file mode 100644 index 000000000..b86612ddd --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/_registry.py @@ -0,0 +1,111 @@ +"""Adapter registry for guardrail scope detection and target wrapping.""" + +from typing import Any, Protocol, runtime_checkable + +from uipath.core.guardrails import GuardrailScope + +from ._core import _EvaluatorFn +from ._enums import GuardrailExecutionStage +from ._models import GuardrailAction + + +@runtime_checkable +class GuardrailTargetAdapter(Protocol): + """Protocol for framework-specific guardrail adapters. + + Implement this protocol to teach :func:`guardrail` how to handle objects + from a particular framework. Register instances via + :func:`register_guardrail_adapter`. + """ + + def detect_scope(self, target: Any) -> GuardrailScope | None: + """Return the guardrail scope for *target*, or ``None`` if not handled. + + Args: + target: Object being decorated. + + Returns: + :class:`~uipath.core.guardrails.GuardrailScope` if recognised, ``None`` otherwise. + """ + ... + + def wrap( + self, + target: Any, + evaluator: _EvaluatorFn, + action: GuardrailAction, + name: str, + stage: GuardrailExecutionStage, + ) -> Any: + """Wrap *target* with guardrail enforcement logic. + + Args: + target: Object to wrap. + evaluator: Unified evaluation callable from :func:`_make_evaluator`. + action: Action to invoke on validation failure. + name: Human-readable guardrail name. + stage: When to evaluate (PRE, POST, or PRE_AND_POST). + + Returns: + Wrapped object, same type or duck-type compatible. + """ + ... + + +# Module-level registry. Later-registered adapters take priority (inserted at 0). +_adapters: list[GuardrailTargetAdapter] = [] + + +def register_guardrail_adapter(adapter: GuardrailTargetAdapter) -> None: + """Register a framework adapter for the ``@guardrail`` decorator. + + Later-registered adapters are tried first. + + Args: + adapter: An instance implementing :class:`GuardrailTargetAdapter`. + """ + _adapters.insert(0, adapter) + + +def detect_scope_from_adapters(target: Any) -> GuardrailScope | None: + """Ask registered adapters to identify the scope of *target*. + + Returns the first non-``None`` result, or ``None`` if no adapter recognises + the target. + + Args: + target: The object being decorated. + + Returns: + :class:`~uipath.core.guardrails.GuardrailScope` or ``None``. + """ + for adapter in _adapters: + scope = adapter.detect_scope(target) + if scope is not None: + return scope + return None + + +def wrap_with_adapter( + target: Any, + evaluator: _EvaluatorFn, + action: GuardrailAction, + name: str, + stage: GuardrailExecutionStage, +) -> Any: + """Ask the first matching adapter to wrap *target*. + + Args: + target: The object to wrap. + evaluator: Unified evaluation callable. + action: Action on violation. + name: Guardrail name. + stage: Execution stage. + + Returns: + Wrapped object, or *target* unchanged if no adapter handles it. + """ + for adapter in _adapters: + if adapter.detect_scope(target) is not None: + return adapter.wrap(target, evaluator, action, name, stage) + return target diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/__init__.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/__init__.py new file mode 100644 index 000000000..159b5052d --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/__init__.py @@ -0,0 +1,14 @@ +"""Guardrail validators for the ``@guardrail`` decorator.""" + +from ._base import GuardrailValidatorBase +from .custom import CustomValidator, RuleFunction +from .pii import PIIValidator +from .prompt_injection import PromptInjectionValidator + +__all__ = [ + "GuardrailValidatorBase", + "PIIValidator", + "PromptInjectionValidator", + "CustomValidator", + "RuleFunction", +] diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/_base.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/_base.py new file mode 100644 index 000000000..0eedbd55b --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/_base.py @@ -0,0 +1,103 @@ +"""Abstract base class for guardrail validators.""" + +from typing import Any, ClassVar + +from uipath.core.guardrails import GuardrailScope, GuardrailValidationResult + +from uipath.platform.guardrails.guardrails import BuiltInValidatorGuardrail + +from .._enums import GuardrailExecutionStage + + +class GuardrailValidatorBase: + """Base class for guardrail validators. + + Subclasses implement :meth:`get_built_in_guardrail` for API-based validation + or :meth:`evaluate` for local Python-based validation. + """ + + supported_scopes: ClassVar[list[GuardrailScope]] = [] + """Scopes this validator supports. Empty list means all scopes are allowed.""" + + supported_stages: ClassVar[list[GuardrailExecutionStage]] = [] + """Stages this validator supports. Empty list means all stages are allowed.""" + + def get_built_in_guardrail( + self, + scope: GuardrailScope, + name: str, + description: str | None, + enabled_for_evals: bool, + ) -> BuiltInValidatorGuardrail | None: + """Build a UiPath API guardrail instance for this validator. + + Override in API-based validators (e.g. :class:`PIIValidator`). Returns + ``None`` by default, causing :meth:`evaluate` to be used instead. + + Args: + scope: Resolved scope of the decorated object. + name: Name for the guardrail. + description: Optional description. + enabled_for_evals: Whether active in evaluation scenarios. + + Returns: + :class:`BuiltInValidatorGuardrail` or ``None``. + """ + return None + + def evaluate( + self, + data: str | dict[str, Any], + stage: GuardrailExecutionStage, + input_data: dict[str, Any] | None, + output_data: dict[str, Any] | None, + ) -> GuardrailValidationResult: + """Perform local validation without a UiPath API call. + + Override in local validators (e.g. :class:`CustomValidator`). Only called + when :meth:`get_built_in_guardrail` returns ``None``. + + Args: + data: Primary data being evaluated. + stage: Current execution stage (PRE or POST). + input_data: Normalised input dict, or ``None`` if unavailable. + output_data: Normalised output dict, or ``None`` at PRE stage. + + Returns: + :class:`~uipath.core.guardrails.GuardrailValidationResult` with + PASSED or VALIDATION_FAILED. + """ + raise NotImplementedError( + f"{type(self).__name__} must implement either get_built_in_guardrail() " + "for API-based validation or evaluate() for local validation." + ) + + def validate_scope(self, scope: GuardrailScope) -> None: + """Raise ``ValueError`` if *scope* is not in :attr:`supported_scopes`. + + Args: + scope: Resolved scope of the decorated object. + + Raises: + ValueError: If :attr:`supported_scopes` is non-empty and *scope* is absent. + """ + if self.supported_scopes and scope not in self.supported_scopes: + raise ValueError( + f"{type(self).__name__} does not support scope {scope!r}. " + f"Supported scopes: {[s.value for s in self.supported_scopes]}" + ) + + def validate_stage(self, stage: GuardrailExecutionStage) -> None: + """Raise ``ValueError`` if *stage* is not in :attr:`supported_stages`. + + Args: + stage: Requested execution stage. + + Raises: + ValueError: If :attr:`supported_stages` is non-empty and *stage* is absent. + """ + if self.supported_stages and stage not in self.supported_stages: + raise ValueError( + f"{type(self).__name__} does not support stage {stage!r}. " + f"Supported stages: {[s.value for s in self.supported_stages]}" + ) diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/custom.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/custom.py new file mode 100644 index 000000000..ea1e0e67a --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/custom.py @@ -0,0 +1,106 @@ +"""Custom (rule-based) guardrail validator.""" + +import inspect +from typing import Any, Callable, ClassVar + +from uipath.core.guardrails import ( + GuardrailScope, + GuardrailValidationResult, + GuardrailValidationResultType, +) + +from .._enums import GuardrailExecutionStage +from ._base import GuardrailValidatorBase + +RuleFunction = ( + Callable[[dict[str, Any]], bool] | Callable[[dict[str, Any], dict[str, Any]], bool] +) +"""Type alias for custom rule functions passed to :class:`CustomValidator`. + +A rule returns ``True`` to signal a violation, ``False`` to pass. It accepts +either one parameter (the input or output dict) or two parameters +(input dict, output dict — POST stage only). +""" + + +class CustomValidator(GuardrailValidatorBase): + """Validate tool input/output using a local Python rule function. + + No UiPath API call is made. Restricted to TOOL scope only. + + Args: + rule: A :data:`RuleFunction` returning ``True`` to signal a violation. + Must accept 1 or 2 parameters. + + Raises: + ValueError: If *rule* is not callable or has an unsupported parameter count. + """ + + supported_scopes = [GuardrailScope.TOOL] + supported_stages: ClassVar[list[GuardrailExecutionStage]] = [] # all stages allowed + + def __init__(self, rule: RuleFunction) -> None: + """Initialize CustomValidator with a rule callable.""" + if not callable(rule): + raise ValueError(f"rule must be callable, got {type(rule)}") + sig = inspect.signature(rule) + param_count = len(sig.parameters) + if param_count not in (1, 2): + raise ValueError(f"rule must have 1 or 2 parameters, got {param_count}") + self.rule = rule + self._param_count = param_count + + def evaluate( + self, + data: str | dict[str, Any], + stage: GuardrailExecutionStage, + input_data: dict[str, Any] | None, + output_data: dict[str, Any] | None, + ) -> GuardrailValidationResult: + """Evaluate the rule against the tool input or output dict. + + Args: + data: Unused; the rule operates on *input_data* or *output_data*. + stage: Current stage (PRE or POST). + input_data: Normalised tool input dict. + output_data: Normalised tool output dict, or ``None`` at PRE stage. + + Returns: + :class:`~uipath.core.guardrails.GuardrailValidationResult` with + PASSED or VALIDATION_FAILED. + """ + try: + if self._param_count == 2: + # Two-parameter rules require both dicts — POST stage only. + if input_data is None or output_data is None: + return GuardrailValidationResult( + result=GuardrailValidationResultType.PASSED, + reason="Two-parameter rule skipped: input or output data unavailable", + ) + violation = self.rule(input_data, output_data) # type: ignore[call-arg] + else: + # One-parameter rules use input_data at PRE, output_data at POST. + target = ( + input_data if stage == GuardrailExecutionStage.PRE else output_data + ) + if target is None: + return GuardrailValidationResult( + result=GuardrailValidationResultType.PASSED, + reason="Rule skipped: data unavailable at this stage", + ) + violation = self.rule(target) # type: ignore[call-arg] + except Exception as exc: + return GuardrailValidationResult( + result=GuardrailValidationResultType.PASSED, + reason=f"Rule raised exception: {exc}", + ) + + if violation: + return GuardrailValidationResult( + result=GuardrailValidationResultType.VALIDATION_FAILED, + reason="Rule detected violation", + ) + return GuardrailValidationResult( + result=GuardrailValidationResultType.PASSED, + reason="Rule passed", + ) diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/pii.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/pii.py new file mode 100644 index 000000000..203f37c79 --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/pii.py @@ -0,0 +1,83 @@ +"""PII detection guardrail validator.""" + +from typing import Any, Sequence +from uuid import uuid4 + +from uipath.core.guardrails import GuardrailScope, GuardrailSelector + +from uipath.platform.guardrails.guardrails import ( + BuiltInValidatorGuardrail, + EnumListParameterValue, + MapEnumParameterValue, +) + +from .._models import PIIDetectionEntity +from ._base import GuardrailValidatorBase + + +class PIIValidator(GuardrailValidatorBase): + """Validate data for PII entities using the UiPath PII detection API. + + Supported at all scopes and stages. + + Args: + entities: One or more :class:`~uipath.platform.guardrails.decorators.PIIDetectionEntity` + instances specifying which PII types to detect and their confidence thresholds. + + Raises: + ValueError: If *entities* is empty. + """ + + # All scopes and stages supported — inherits empty lists from base (unrestricted). + + def __init__(self, entities: Sequence[PIIDetectionEntity]) -> None: + """Initialize PIIValidator with a list of entities to detect.""" + if not entities: + raise ValueError("entities must be provided and non-empty") + self.entities = list(entities) + + def get_built_in_guardrail( + self, + scope: GuardrailScope, + name: str, + description: str | None, + enabled_for_evals: bool, + ) -> BuiltInValidatorGuardrail: + """Build a PII detection :class:`BuiltInValidatorGuardrail`. + + Args: + scope: Resolved scope of the decorated object. + name: Name for the guardrail. + description: Optional description. + enabled_for_evals: Whether active in evaluation scenarios. + + Returns: + Configured :class:`BuiltInValidatorGuardrail` for PII detection. + """ + entity_names = [entity.name for entity in self.entities] + entity_thresholds: dict[str, Any] = { + entity.name: entity.threshold for entity in self.entities + } + + return BuiltInValidatorGuardrail( + id=str(uuid4()), + name=name, + description=description + or f"Detects PII entities: {', '.join(entity_names)}", + enabled_for_evals=enabled_for_evals, + selector=GuardrailSelector(scopes=[scope]), + guardrail_type="builtInValidator", + validator_type="pii_detection", + validator_parameters=[ + EnumListParameterValue( + parameter_type="enum-list", + id="entities", + value=entity_names, + ), + MapEnumParameterValue( + parameter_type="map-enum", + id="entityThresholds", + value=entity_thresholds, + ), + ], + ) diff --git a/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/prompt_injection.py b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/prompt_injection.py new file mode 100644 index 000000000..a9fecd1ca --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/guardrails/decorators/validators/prompt_injection.py @@ -0,0 +1,71 @@ +"""Prompt injection detection guardrail validator.""" + +from uuid import uuid4 + +from uipath.core.guardrails import GuardrailScope, GuardrailSelector + +from uipath.platform.guardrails.guardrails import ( + BuiltInValidatorGuardrail, + NumberParameterValue, +) + +from .._enums import GuardrailExecutionStage +from ._base import GuardrailValidatorBase + + +class PromptInjectionValidator(GuardrailValidatorBase): + """Validate LLM input for prompt injection attacks via the UiPath API. + + Restricted to LLM scope and PRE stage only. + + Args: + threshold: Detection confidence threshold (0.0–1.0). Defaults to ``0.5``. + + Raises: + ValueError: If *threshold* is outside [0.0, 1.0]. + """ + + supported_scopes = [GuardrailScope.LLM] + supported_stages = [GuardrailExecutionStage.PRE] + + def __init__(self, threshold: float = 0.5) -> None: + """Initialize PromptInjectionValidator with a detection threshold.""" + if not 0.0 <= threshold <= 1.0: + raise ValueError(f"threshold must be between 0.0 and 1.0, got {threshold}") + self.threshold = threshold + + def get_built_in_guardrail( + self, + scope: GuardrailScope, + name: str, + description: str | None, + enabled_for_evals: bool, + ) -> BuiltInValidatorGuardrail: + """Build a prompt injection :class:`BuiltInValidatorGuardrail`. + + Args: + scope: Resolved scope (must be LLM). + name: Name for the guardrail. + description: Optional description. + enabled_for_evals: Whether active in evaluation scenarios. + + Returns: + Configured :class:`BuiltInValidatorGuardrail` for prompt injection. + """ + return BuiltInValidatorGuardrail( + id=str(uuid4()), + name=name, + description=description + or f"Detects prompt injection with threshold {self.threshold}", + enabled_for_evals=enabled_for_evals, + selector=GuardrailSelector(scopes=[GuardrailScope.LLM]), + guardrail_type="builtInValidator", + validator_type="prompt_injection", + validator_parameters=[ + NumberParameterValue( + parameter_type="number", + id="threshold", + value=self.threshold, + ), + ], + ) diff --git a/packages/uipath-platform/tests/services/test_guardrails_decorators.py b/packages/uipath-platform/tests/services/test_guardrails_decorators.py new file mode 100644 index 000000000..a77460eed --- /dev/null +++ b/packages/uipath-platform/tests/services/test_guardrails_decorators.py @@ -0,0 +1,895 @@ +"""Tests for the guardrails decorator framework in uipath-platform. + +Focus: meaningful business behaviour — constraint enforcement, routing logic, +API integration, and end-to-end decorator scenarios modelled on the +joke-agent-decorator sample. Attribute-access and isinstance checks have +been intentionally omitted. +""" + +from __future__ import annotations + +import logging +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from uipath.core.guardrails import ( + GuardrailScope, + GuardrailValidationResult, + GuardrailValidationResultType, +) + +from uipath.platform.guardrails.decorators import ( + BlockAction, + CustomValidator, + GuardrailAction, + GuardrailBlockException, + GuardrailExecutionStage, + GuardrailValidatorBase, + LogAction, + LoggingSeverityLevel, + PIIDetectionEntity, + PIIDetectionEntityType, + PIIValidator, + PromptInjectionValidator, + guardrail, + register_guardrail_adapter, +) +from uipath.platform.guardrails.decorators._core import ( + _evaluate_rules, + _extract_input, + _extract_output, + _make_evaluator, + _rewrap_input, +) +from uipath.platform.guardrails.decorators._registry import ( + _adapters, + detect_scope_from_adapters, + wrap_with_adapter, +) + +# --------------------------------------------------------------------------- +# Shared result constants +# --------------------------------------------------------------------------- + +_PASSED = GuardrailValidationResult( + result=GuardrailValidationResultType.PASSED, + reason="ok", +) +_FAILED = GuardrailValidationResult( + result=GuardrailValidationResultType.VALIDATION_FAILED, + reason="violation detected", +) + + +# --------------------------------------------------------------------------- +# Registry isolation fixture +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _isolate_adapter_registry(): + """Snapshot and restore the global adapter registry around every test.""" + snapshot = list(_adapters) + yield + _adapters.clear() + _adapters.extend(snapshot) + + +# --------------------------------------------------------------------------- +# Minimal fake types for adapter tests (no LangChain dependency) +# --------------------------------------------------------------------------- + + +class _DummyTarget: + """Minimal callable target recognised by _DummyAdapter.""" + + def __init__(self, return_value: Any = None) -> None: + self.return_value = ( + return_value if return_value is not None else {"output": "result"} + ) + self.invoke_calls: list[Any] = [] + + def invoke(self, args: Any) -> Any: + self.invoke_calls.append(args) + return self.return_value + + +class _WrappedDummyTarget: + """A _DummyTarget wrapped with guardrail evaluation.""" + + def __init__( + self, + target: Any, + evaluator: Any, + action: GuardrailAction, + name: str, + stage: GuardrailExecutionStage, + ) -> None: + self._target = target + self._evaluator = evaluator + self._action = action + self._name = name + self._stage = stage + + def invoke(self, args: Any) -> Any: + input_data = args if isinstance(args, dict) else {"input": args} + if self._stage in ( + GuardrailExecutionStage.PRE, + GuardrailExecutionStage.PRE_AND_POST, + ): + result = self._evaluator( + input_data, GuardrailExecutionStage.PRE, input_data, None + ) + if result.result == GuardrailValidationResultType.VALIDATION_FAILED: + self._action.handle_validation_result(result, input_data, self._name) + raw = self._target.invoke(args) + output_data = raw if isinstance(raw, dict) else {"output": raw} + if self._stage in ( + GuardrailExecutionStage.POST, + GuardrailExecutionStage.PRE_AND_POST, + ): + result = self._evaluator( + output_data, GuardrailExecutionStage.POST, input_data, output_data + ) + if result.result == GuardrailValidationResultType.VALIDATION_FAILED: + self._action.handle_validation_result(result, output_data, self._name) + return raw + + +class _DummyAdapter: + """Adapter that handles _DummyTarget instances at TOOL scope.""" + + def detect_scope(self, target: Any) -> GuardrailScope | None: + if isinstance(target, (_DummyTarget, _WrappedDummyTarget)): + return GuardrailScope.TOOL + return None + + def wrap( + self, + target: Any, + evaluator: Any, + action: GuardrailAction, + name: str, + stage: GuardrailExecutionStage, + ) -> Any: + return _WrappedDummyTarget(target, evaluator, action, name, stage) + + +# --------------------------------------------------------------------------- +# 1. PIIDetectionEntity — threshold boundary enforcement +# --------------------------------------------------------------------------- + + +class TestPIIDetectionEntity: + def test_threshold_below_zero_raises(self): + with pytest.raises(ValueError, match="0.0 and 1.0"): + PIIDetectionEntity(name="Email", threshold=-0.1) + + def test_threshold_above_one_raises(self): + with pytest.raises(ValueError, match="0.0 and 1.0"): + PIIDetectionEntity(name="Email", threshold=1.1) + + +# --------------------------------------------------------------------------- +# 2. LogAction — does NOT stop execution; uses configured severity +# --------------------------------------------------------------------------- + + +class TestLogAction: + def test_violation_logs_guardrail_name_and_execution_continues(self, caplog): + action = LogAction() + with caplog.at_level(logging.WARNING): + result = action.handle_validation_result(_FAILED, "data", "MyGuardrail") + assert result is None # execution continues + assert any("MyGuardrail" in r.message for r in caplog.records) + + def test_pass_emits_no_log(self, caplog): + action = LogAction() + with caplog.at_level(logging.WARNING): + action.handle_validation_result(_PASSED, "data", "G") + assert not caplog.records + + def test_custom_message_overrides_reason(self, caplog): + action = LogAction(message="custom alert") + with caplog.at_level(logging.WARNING): + action.handle_validation_result(_FAILED, "data", "G") + assert any("custom alert" in r.message for r in caplog.records) + + def test_default_message_includes_validation_reason(self, caplog): + action = LogAction() + with caplog.at_level(logging.WARNING): + action.handle_validation_result(_FAILED, "data", "G") + assert any("violation detected" in r.message for r in caplog.records) + + def test_severity_level_respected(self, caplog): + action = LogAction(severity_level=LoggingSeverityLevel.ERROR) + with caplog.at_level(logging.ERROR): + action.handle_validation_result(_FAILED, "data", "G") + assert any(r.levelno == logging.ERROR for r in caplog.records) + + +# --------------------------------------------------------------------------- +# 3. BlockAction — halts execution with configurable title/detail +# --------------------------------------------------------------------------- + + +class TestBlockAction: + def test_pass_does_not_raise(self): + assert BlockAction().handle_validation_result(_PASSED, "data", "G") is None + + def test_violation_raises_block_exception(self): + with pytest.raises(GuardrailBlockException): + BlockAction().handle_validation_result(_FAILED, "data", "G") + + def test_default_title_includes_guardrail_name(self): + with pytest.raises(GuardrailBlockException) as exc_info: + BlockAction().handle_validation_result(_FAILED, "data", "MyGuardrail") + assert "MyGuardrail" in exc_info.value.title + + def test_default_detail_uses_validation_reason(self): + with pytest.raises(GuardrailBlockException) as exc_info: + BlockAction().handle_validation_result(_FAILED, "data", "G") + assert exc_info.value.detail == "violation detected" + + def test_custom_title_and_detail_take_precedence(self): + with pytest.raises(GuardrailBlockException) as exc_info: + BlockAction( + title="Joke too long", detail="Joke > 1000 chars" + ).handle_validation_result(_FAILED, "data", "G") + assert exc_info.value.title == "Joke too long" + assert exc_info.value.detail == "Joke > 1000 chars" + + def test_empty_reason_falls_back_to_generic_detail(self): + failed_empty = GuardrailValidationResult( + result=GuardrailValidationResultType.VALIDATION_FAILED, + reason="", + ) + with pytest.raises(GuardrailBlockException) as exc_info: + BlockAction().handle_validation_result(failed_empty, "data", "G") + assert exc_info.value.detail # not empty — fallback applied + + +# --------------------------------------------------------------------------- +# 4. GuardrailValidatorBase — scope/stage enforcement contracts +# --------------------------------------------------------------------------- + + +class TestGuardrailValidatorBase: + class _ToolOnlyValidator(GuardrailValidatorBase): + supported_scopes = [GuardrailScope.TOOL] + + class _PreOnlyValidator(GuardrailValidatorBase): + supported_stages = [GuardrailExecutionStage.PRE] + + class _UnrestrictedValidator(GuardrailValidatorBase): + pass + + def test_unsupported_scope_raises(self): + with pytest.raises(ValueError, match="does not support scope"): + self._ToolOnlyValidator().validate_scope(GuardrailScope.LLM) + + def test_unsupported_stage_raises(self): + with pytest.raises(ValueError, match="does not support stage"): + self._PreOnlyValidator().validate_stage(GuardrailExecutionStage.POST) + + def test_get_built_in_guardrail_returns_none_by_default(self): + assert ( + self._UnrestrictedValidator().get_built_in_guardrail( + GuardrailScope.TOOL, "g", None, True + ) + is None + ) + + def test_evaluate_raises_not_implemented(self): + with pytest.raises(NotImplementedError): + self._UnrestrictedValidator().evaluate( + "data", GuardrailExecutionStage.PRE, {}, None + ) + + +# --------------------------------------------------------------------------- +# 5. PIIValidator — API guardrail construction +# --------------------------------------------------------------------------- + + +class TestPIIValidator: + def test_empty_entities_raises(self): + with pytest.raises(ValueError, match="entities"): + PIIValidator(entities=[]) + + def test_builds_pii_detection_guardrail(self): + v = PIIValidator( + entities=[PIIDetectionEntity(PIIDetectionEntityType.EMAIL, 0.5)] + ) + g = v.get_built_in_guardrail(GuardrailScope.TOOL, "PII", None, True) + assert g is not None + assert g.validator_type == "pii_detection" + assert g.name == "PII" + + def test_entity_names_and_thresholds_in_api_parameters(self): + v = PIIValidator( + entities=[ + PIIDetectionEntity(PIIDetectionEntityType.EMAIL, 0.6), + PIIDetectionEntity(PIIDetectionEntityType.PERSON, 0.8), + ] + ) + g = v.get_built_in_guardrail(GuardrailScope.TOOL, "G", None, True) + param_by_id = {p.id: p for p in g.validator_parameters} + entities_value = param_by_id["entities"].value + assert isinstance(entities_value, list) + assert "Email" in entities_value + assert "Person" in entities_value + thresholds_value = param_by_id["entityThresholds"].value + assert isinstance(thresholds_value, dict) + assert thresholds_value["Email"] == 0.6 + assert thresholds_value["Person"] == 0.8 + + +# --------------------------------------------------------------------------- +# 6. PromptInjectionValidator — LLM-only, PRE-only, threshold validation +# --------------------------------------------------------------------------- + + +class TestPromptInjectionValidator: + def test_threshold_below_zero_raises(self): + with pytest.raises(ValueError, match="threshold"): + PromptInjectionValidator(threshold=-0.1) + + def test_threshold_above_one_raises(self): + with pytest.raises(ValueError, match="threshold"): + PromptInjectionValidator(threshold=1.1) + + def test_restricted_to_llm_scope_only(self): + v = PromptInjectionValidator() + v.validate_scope(GuardrailScope.LLM) # ok + with pytest.raises(ValueError): + v.validate_scope(GuardrailScope.TOOL) + + def test_restricted_to_pre_stage_only(self): + v = PromptInjectionValidator() + v.validate_stage(GuardrailExecutionStage.PRE) # ok + with pytest.raises(ValueError): + v.validate_stage(GuardrailExecutionStage.POST) + + def test_builds_prompt_injection_guardrail_with_threshold(self): + v = PromptInjectionValidator(threshold=0.7) + g = v.get_built_in_guardrail(GuardrailScope.LLM, "PI", None, True) + assert g.validator_type == "prompt_injection" + threshold_param = next(p for p in g.validator_parameters if p.id == "threshold") + assert threshold_param.value == 0.7 + + +# --------------------------------------------------------------------------- +# 7. CustomValidator — rule routing and error handling +# --------------------------------------------------------------------------- + + +class TestCustomValidator: + def test_non_callable_raises(self): + with pytest.raises(ValueError, match="callable"): + CustomValidator(rule="not_a_function") # type: ignore[arg-type] + + def test_wrong_arity_raises(self): + with pytest.raises(ValueError, match="1 or 2"): + CustomValidator(rule=lambda: True) # type: ignore[arg-type] + with pytest.raises(ValueError, match="1 or 2"): + CustomValidator(rule=lambda a, b, c: True) # type: ignore[arg-type] + + def test_restricted_to_tool_scope(self): + v = CustomValidator(rule=lambda args: False) + with pytest.raises(ValueError): + v.validate_scope(GuardrailScope.LLM) + + def test_one_param_pre_receives_input_data(self): + received: list[Any] = [] + + def capture_pre(args: dict[str, Any]) -> bool: + received.append(args) + return False + + CustomValidator(rule=capture_pre).evaluate( + {}, GuardrailExecutionStage.PRE, {"a": 1}, None + ) + assert received == [{"a": 1}] + + def test_one_param_post_receives_output_data(self): + received: list[Any] = [] + + def capture_post(args: dict[str, Any]) -> bool: + received.append(args) + return False + + CustomValidator(rule=capture_post).evaluate( + {}, GuardrailExecutionStage.POST, {"in": 1}, {"out": 2} + ) + assert received == [{"out": 2}] + + def test_two_param_post_receives_input_and_output(self): + received: list[Any] = [] + + def rule(inp, out): + received.append((inp, out)) + return False + + CustomValidator(rule=rule).evaluate( + {}, GuardrailExecutionStage.POST, {"in": 1}, {"out": 2} + ) + assert received == [({"in": 1}, {"out": 2})] + + def test_two_param_rule_skipped_when_input_missing(self): + result = CustomValidator(rule=lambda a, b: True).evaluate( + {}, GuardrailExecutionStage.POST, None, {"out": 2} + ) + assert result.result == GuardrailValidationResultType.PASSED + + def test_rule_exception_is_caught_and_reported_as_passed(self): + def bad_rule(args): + raise RuntimeError("oops") + + result = CustomValidator(rule=bad_rule).evaluate( + {}, GuardrailExecutionStage.PRE, {"x": 1}, None + ) + assert result.result == GuardrailValidationResultType.PASSED + assert "oops" in result.reason + + +# --------------------------------------------------------------------------- +# 8. LangGraph tool-call envelope: extract and rewrap +# --------------------------------------------------------------------------- + + +class TestToolCallEnvelopeHandling: + """_extract_input unwraps LangGraph envelopes; _rewrap_input re-wraps without mutation.""" + + def test_extract_unwraps_langgraph_envelope(self): + envelope = { + "type": "tool_call", + "args": {"joke": "hello"}, + "name": "t", + "id": "1", + } + assert _extract_input(envelope) == {"joke": "hello"} + + def test_extract_non_dict_args_falls_through_to_outer_dict(self): + envelope = {"type": "tool_call", "args": "string_args"} + result = _extract_input(envelope) + assert "type" in result # not unwrapped + + def test_rewrap_restores_envelope_structure(self): + original = {"type": "tool_call", "args": {"joke": "orig"}, "id": "1"} + result = _rewrap_input(original, {"joke": "modified"}) + assert result["args"] == {"joke": "modified"} + assert result["type"] == "tool_call" + + def test_rewrap_does_not_mutate_original(self): + original = {"type": "tool_call", "args": {"joke": "orig"}, "id": "1"} + _rewrap_input(original, {"joke": "new"}) + assert original["args"] == {"joke": "orig"} + + +# --------------------------------------------------------------------------- +# 9. _extract_output — normalises tool output for rule evaluation +# --------------------------------------------------------------------------- + + +class TestExtractOutput: + def test_json_string_parsed_to_dict(self): + assert _extract_output('{"key": "val"}') == {"key": "val"} + + def test_json_array_string_wrapped(self): + assert _extract_output('["a", "b"]') == {"output": ["a", "b"]} + + def test_plain_string_wrapped(self): + assert _extract_output("hello") == {"output": "hello"} + + def test_python_literal_dict_string_parsed(self): + assert _extract_output("{'key': 'val'}") == {"key": "val"} + + +# --------------------------------------------------------------------------- +# 10. _evaluate_rules — non-obvious evaluation semantics +# --------------------------------------------------------------------------- + + +class TestEvaluateRules: + def test_empty_rules_always_trigger_action(self): + result = _evaluate_rules([], GuardrailExecutionStage.PRE, {}, None) + assert result.result == GuardrailValidationResultType.VALIDATION_FAILED + assert "Empty rules" in result.reason + + def test_single_violation_fails(self): + result = _evaluate_rules( + [lambda args: True], GuardrailExecutionStage.PRE, {"x": 1}, None + ) + assert result.result == GuardrailValidationResultType.VALIDATION_FAILED + + def test_any_one_passing_rule_makes_result_pass(self): + """Semantics: one passing rule is sufficient to pass overall.""" + rules = [lambda args: True, lambda args: False] + result = _evaluate_rules(rules, GuardrailExecutionStage.PRE, {"x": 1}, None) + assert result.result == GuardrailValidationResultType.PASSED + + def test_all_rules_must_violate_to_fail(self): + rules = [lambda args: True, lambda args: True] + result = _evaluate_rules(rules, GuardrailExecutionStage.PRE, {"x": 1}, None) + assert result.result == GuardrailValidationResultType.VALIDATION_FAILED + + def test_two_param_rule_skipped_at_pre_stage(self): + """Two-param rules need both input+output — silently skipped at PRE.""" + result = _evaluate_rules( + [lambda a, b: True], GuardrailExecutionStage.PRE, {"x": 1}, None + ) + assert result.result == GuardrailValidationResultType.PASSED + assert "No applicable rules" in result.reason + + def test_post_one_param_rule_receives_output(self): + received: list[Any] = [] + + def capture_out(out: dict[str, Any]) -> bool: + received.append(out) + return False + + _evaluate_rules( + [capture_out], + GuardrailExecutionStage.POST, + {"in": 1}, + {"out": 2}, + ) + assert received == [{"out": 2}] + + def test_post_two_param_rule_receives_both(self): + received: list[Any] = [] + + def rule(inp, out): + received.append((inp, out)) + return False + + _evaluate_rules([rule], GuardrailExecutionStage.POST, {"in": 1}, {"out": 2}) + assert received == [({"in": 1}, {"out": 2})] + + def test_rule_exception_counts_as_violation(self): + def bad(args): + raise ValueError("boom") + + result = _evaluate_rules([bad], GuardrailExecutionStage.PRE, {"x": 1}, None) + assert result.result == GuardrailValidationResultType.VALIDATION_FAILED + + def test_no_applicable_rules_returns_passed(self): + """All rules are two-param but we're at PRE — none applicable → PASSED.""" + rules = [lambda a, b: True, lambda a, b: False] + result = _evaluate_rules(rules, GuardrailExecutionStage.PRE, {"x": 1}, None) + assert result.result == GuardrailValidationResultType.PASSED + assert "No applicable rules" in result.reason + + +# --------------------------------------------------------------------------- +# 11. _make_evaluator — local vs API path, lazy UiPath instantiation +# --------------------------------------------------------------------------- + + +class TestMakeEvaluator: + def test_local_path_delegates_to_validator_evaluate(self): + mock_validator = MagicMock() + mock_validator.evaluate.return_value = _PASSED + evaluator = _make_evaluator(mock_validator, built_in_guardrail=None) + result = evaluator({"data": 1}, GuardrailExecutionStage.PRE, {"a": 1}, None) + mock_validator.evaluate.assert_called_once_with( + {"data": 1}, GuardrailExecutionStage.PRE, {"a": 1}, None + ) + assert result == _PASSED + + def test_api_path_calls_uipath_guardrails(self): + mock_uipath = MagicMock() + mock_uipath.guardrails.evaluate_guardrail.return_value = _PASSED + with patch("uipath.platform.UiPath", return_value=mock_uipath): + evaluator = _make_evaluator(MagicMock(), built_in_guardrail=MagicMock()) + result = evaluator("text", GuardrailExecutionStage.PRE, None, None) + assert result == _PASSED + + def test_uipath_created_lazily_and_reused(self): + """UiPath() must not be instantiated at decoration time — only on first call.""" + with patch("uipath.platform.UiPath") as MockUiPath: + MockUiPath.return_value.guardrails.evaluate_guardrail.return_value = _PASSED + evaluator = _make_evaluator(MagicMock(), built_in_guardrail=MagicMock()) + assert MockUiPath.call_count == 0 + evaluator("text", GuardrailExecutionStage.PRE, None, None) + assert MockUiPath.call_count == 1 + evaluator("text", GuardrailExecutionStage.PRE, None, None) + assert MockUiPath.call_count == 1 # reused, not re-created + + +# --------------------------------------------------------------------------- +# 12. Adapter registry — priority, scope detection, delegation +# --------------------------------------------------------------------------- + + +class TestAdapterRegistry: + def test_last_registered_adapter_has_highest_priority(self): + a = _DummyAdapter() + b = _DummyAdapter() + register_guardrail_adapter(a) + register_guardrail_adapter(b) + assert _adapters[0] is b + + def test_first_matching_adapter_wins(self): + class _AgentAdapter: + def detect_scope(self, t: Any) -> GuardrailScope | None: + return GuardrailScope.AGENT if isinstance(t, _DummyTarget) else None + + def wrap(self, *a: Any, **kw: Any) -> Any: + return None + + register_guardrail_adapter(_DummyAdapter()) # TOOL + register_guardrail_adapter(_AgentAdapter()) # AGENT (higher priority) + assert detect_scope_from_adapters(_DummyTarget()) == GuardrailScope.AGENT + + def test_unrecognised_target_returns_none(self): + register_guardrail_adapter(_DummyAdapter()) + assert detect_scope_from_adapters("plain_string") is None + + def test_wrap_delegates_to_matching_adapter(self): + register_guardrail_adapter(_DummyAdapter()) + wrapped = wrap_with_adapter( + _DummyTarget(), + MagicMock(return_value=_PASSED), + LogAction(), + "G", + GuardrailExecutionStage.PRE, + ) + assert isinstance(wrapped, _WrappedDummyTarget) + + +# --------------------------------------------------------------------------- +# 13. @guardrail decorator — validation and dispatch +# --------------------------------------------------------------------------- + + +class TestGuardrailDecorator: + def test_missing_action_raises(self): + with pytest.raises(ValueError, match="action must be provided"): + guardrail( + _DummyTarget(), + validator=CustomValidator(lambda args: False), + action=None, # type: ignore[arg-type] + ) + + def test_non_action_instance_raises(self): + register_guardrail_adapter(_DummyAdapter()) + with pytest.raises(ValueError, match="GuardrailAction"): + guardrail( + _DummyTarget(), + validator=CustomValidator(lambda args: False), + action="bad", # type: ignore[arg-type] + ) + + def test_invalid_enabled_for_evals_type_raises(self): + register_guardrail_adapter(_DummyAdapter()) + with pytest.raises(ValueError, match="boolean"): + guardrail( + _DummyTarget(), + validator=CustomValidator(lambda args: False), + action=LogAction(), + enabled_for_evals="yes", # type: ignore[arg-type] + ) + + def test_tool_only_validator_on_factory_raises(self): + """CustomValidator (TOOL-only) cannot be applied to a plain factory function.""" + + def create_thing(): + return _DummyTarget() + + with pytest.raises(ValueError, match="BaseTool"): + guardrail( + create_thing, + validator=CustomValidator(lambda args: False), + action=LogAction(), + ) + + def test_scope_violation_caught_at_decoration_time(self): + """PromptInjectionValidator (LLM-only) applied to a TOOL-scope target fails immediately.""" + register_guardrail_adapter(_DummyAdapter()) + with pytest.raises(ValueError, match="does not support scope"): + guardrail( + _DummyTarget(), + validator=PromptInjectionValidator(), + action=LogAction(), + stage=GuardrailExecutionStage.PRE, + ) + + def test_unrecognised_target_without_adapter_raises(self): + class _Opaque: + pass + + with pytest.raises(ValueError, match="cannot be applied"): + guardrail( + _Opaque(), + validator=PIIValidator( + entities=[PIIDetectionEntity(PIIDetectionEntityType.EMAIL)] + ), + action=LogAction(), + ) + + def test_decorator_usage_with_parens(self): + register_guardrail_adapter(_DummyAdapter()) + decorated = guardrail( + validator=CustomValidator(lambda args: False), + action=LogAction(), + stage=GuardrailExecutionStage.PRE, + )(_DummyTarget()) + assert isinstance(decorated, _WrappedDummyTarget) + + def test_factory_callable_wrapped_for_api_validators(self): + pii = PIIValidator(entities=[PIIDetectionEntity(PIIDetectionEntityType.EMAIL)]) + + def create_thing(): + return "something" + + wrapped = guardrail(create_thing, validator=pii, action=LogAction()) + assert callable(wrapped) + + +# --------------------------------------------------------------------------- +# 14. Integration: joke-agent end-to-end scenarios +# --------------------------------------------------------------------------- + + +class TestJokeAgentScenarios: + """Full decoration + invocation scenarios via _DummyAdapter. + + Modelled on the joke-agent-decorator sample — no LangChain dependency. + """ + + @pytest.fixture(autouse=True) + def _register_adapter(self): + register_guardrail_adapter(_DummyAdapter()) + + def _make_tool(self, return_value: Any = None) -> _DummyTarget: + return _DummyTarget(return_value=return_value) + + def test_word_filter_logs_and_tool_still_executes(self): + """LogAction on 'donkey' violation — execution is NOT blocked.""" + target = self._make_tool({"output": "words: 6 letters: 30"}) + wrapped = guardrail( + target, + validator=CustomValidator( + lambda args: "donkey" in args.get("joke", "").lower() + ), + action=LogAction(severity_level=LoggingSeverityLevel.WARNING), + stage=GuardrailExecutionStage.PRE, + name="Joke Content Word Filter", + ) + result = wrapped.invoke({"joke": "Why did the donkey cross the road?"}) + assert result == {"output": "words: 6 letters: 30"} + assert target.invoke_calls # tool was called despite violation + + def test_word_filter_does_not_fire_for_clean_input(self): + target = self._make_tool({"output": "words: 5 letters: 25"}) + wrapped = guardrail( + target, + validator=CustomValidator( + lambda args: "donkey" in args.get("joke", "").lower() + ), + action=LogAction(), + stage=GuardrailExecutionStage.PRE, + name="Joke Content Word Filter", + ) + result = wrapped.invoke({"joke": "Why did the banana split?"}) + assert result == {"output": "words: 5 letters: 25"} + assert len(target.invoke_calls) == 1 + + def test_length_limiter_blocks_before_tool_runs(self): + """BlockAction at PRE — long joke raises and tool is never called.""" + target = self._make_tool() + wrapped = guardrail( + target, + validator=CustomValidator(lambda args: len(args.get("joke", "")) > 1000), + action=BlockAction(title="Joke too long", detail="Joke > 1000 chars"), + stage=GuardrailExecutionStage.PRE, + name="Joke Content Length Limiter", + ) + with pytest.raises(GuardrailBlockException) as exc_info: + wrapped.invoke({"joke": "A" * 1001}) + + assert exc_info.value.title == "Joke too long" + assert exc_info.value.detail == "Joke > 1000 chars" + assert not target.invoke_calls + + def test_length_limiter_passes_short_joke_through(self): + target = self._make_tool({"output": "ok"}) + wrapped = guardrail( + target, + validator=CustomValidator(lambda args: len(args.get("joke", "")) > 1000), + action=BlockAction(title="Joke too long", detail="Joke > 1000 chars"), + stage=GuardrailExecutionStage.PRE, + name="Joke Content Length Limiter", + ) + result = wrapped.invoke({"joke": "Short joke."}) + assert result == {"output": "ok"} + assert len(target.invoke_calls) == 1 + + def test_post_guardrail_fires_after_tool(self): + """POST-stage guardrail — tool runs first, then validator fires.""" + target = self._make_tool({"output": "funny joke"}) + wrapped = guardrail( + target, + validator=CustomValidator(lambda out: True), + action=LogAction(severity_level=LoggingSeverityLevel.WARNING), + stage=GuardrailExecutionStage.POST, + name="Always Filter", + ) + result = wrapped.invoke({"joke": "Any input"}) + assert result == {"output": "funny joke"} + assert len(target.invoke_calls) == 1 + + def test_stacked_guardrails_outer_fires_first(self): + """Outer wrapper's PRE runs before inner — outer exception propagates first.""" + target = self._make_tool({"output": "ok"}) + inner = guardrail( + target, + validator=CustomValidator(lambda args: len(args.get("joke", "")) > 1000), + action=BlockAction(title="Too long", detail="Too long"), + stage=GuardrailExecutionStage.PRE, + name="Length Limiter", + ) + outer = guardrail( + inner, + validator=CustomValidator(lambda args: "donkey" in str(args).lower()), + action=BlockAction(title="Word filter", detail="donkey found"), + stage=GuardrailExecutionStage.PRE, + name="Word Filter", + ) + with pytest.raises(GuardrailBlockException) as exc_info: + outer.invoke({"joke": "donkey joke"}) + assert exc_info.value.title == "Word filter" + + def test_pii_api_block_prevents_tool_execution(self, mock_env_vars): + """PIIValidator routes to UiPath API; FAILED result blocks the tool.""" + target = self._make_tool() + wrapped = guardrail( + target, + validator=PIIValidator( + entities=[PIIDetectionEntity(PIIDetectionEntityType.PERSON, 0.5)] + ), + action=BlockAction(), + stage=GuardrailExecutionStage.PRE, + name="Tool PII Block Detection", + ) + with patch("uipath.platform.UiPath") as MockUiPath: + MockUiPath.return_value.guardrails.evaluate_guardrail.return_value = _FAILED + with pytest.raises(GuardrailBlockException) as exc_info: + wrapped.invoke({"joke": "Hello John Doe"}) + + assert "Tool PII Block Detection" in exc_info.value.title + assert not target.invoke_calls + + def test_pii_api_pass_allows_tool_execution(self, mock_env_vars): + """PIIValidator routes to UiPath API; PASSED result lets the tool run.""" + target = self._make_tool({"output": "ok"}) + wrapped = guardrail( + target, + validator=PIIValidator( + entities=[PIIDetectionEntity(PIIDetectionEntityType.PERSON, 0.5)] + ), + action=BlockAction(), + stage=GuardrailExecutionStage.PRE, + name="Tool PII Detection", + ) + with patch("uipath.platform.UiPath") as MockUiPath: + MockUiPath.return_value.guardrails.evaluate_guardrail.return_value = _PASSED + result = wrapped.invoke({"joke": "Clean input"}) + + assert result == {"output": "ok"} + assert len(target.invoke_calls) == 1 + + def test_prompt_injection_on_tool_scope_raises_at_decoration(self): + """PromptInjectionValidator is LLM-only — decorating a TOOL fails at decoration time.""" + pi = PromptInjectionValidator(threshold=0.5) + with pytest.raises(ValueError, match="does not support scope"): + guardrail( + self._make_tool(), + validator=pi, + action=BlockAction(), + name="PI Check", + stage=GuardrailExecutionStage.PRE, + ) diff --git a/packages/uipath-platform/uv.lock b/packages/uipath-platform/uv.lock index 4360ed46b..0ea4a2f63 100644 --- a/packages/uipath-platform/uv.lock +++ b/packages/uipath-platform/uv.lock @@ -1088,7 +1088,7 @@ dev = [ [[package]] name = "uipath-platform" -version = "0.1.17" +version = "0.1.18" source = { editable = "." } dependencies = [ { name = "httpx" }, diff --git a/packages/uipath/uv.lock b/packages/uipath/uv.lock index 41ecc6a40..77434aaa8 100644 --- a/packages/uipath/uv.lock +++ b/packages/uipath/uv.lock @@ -2682,7 +2682,7 @@ dev = [ [[package]] name = "uipath-platform" -version = "0.1.17" +version = "0.1.18" source = { editable = "../uipath-platform" } dependencies = [ { name = "httpx" },