"""Automated-Decision Guard (G-HUMAN-DEC). For AI use cases configured as requiring human review, workflow/agent results must not automatically trigger defined person-/risk-relevant external effects before the required human review/approval policy is fulfilled. This guard checks the AIUseCaseMetadata of an agent or workflow step and blocks automatic execution of high-risk actions unless approval has been granted. No blanket enforcement for normal CRM automation — only applies to explicitly configured AI use cases. """ from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # Risk levels for automated decisions RISK_LEVELS = { "none": 0, "low": 1, "medium": 2, "high": 3, "critical": 4, } # Actions that require human review when risk_level >= medium HIGH_RISK_ACTIONS = { "send_email", "send_external_message", "delete_entity", "modify_permissions", "execute_payment", "publish_content", "modify_contract", "data_export", } def requires_human_review( ai_use_case_metadata: dict[str, Any] | None, action: str, ) -> bool: """Check whether an action requires human review based on AI use case config. Args: ai_use_case_metadata: The ai_use_case_metadata from AgentDefinition or workflow step config. May contain: - ``risk_level``: none|low|medium|high|critical - ``requires_approval``: bool - ``auto_execute``: bool (if False, always requires review) - ``reviewed_actions``: list of actions that need review action: The action being performed (e.g. "send_email", "delete_entity"). Returns: True if human review is required before the action can execute. """ if not ai_use_case_metadata: return False # No AI use case config → no guard # If auto_execute is explicitly False, always require review if ai_use_case_metadata.get("auto_execute") is False: return True # If requires_approval is True, always require review if ai_use_case_metadata.get("requires_approval") is True: return True # Check risk level risk_level = ai_use_case_metadata.get("risk_level", "none") risk_score = RISK_LEVELS.get(risk_level, 0) # Medium or higher risk always requires review if risk_score >= RISK_LEVELS["medium"]: return True # Check if this specific action is in the reviewed_actions list reviewed_actions = ai_use_case_metadata.get("reviewed_actions", []) if action in reviewed_actions: return True # Check if action is inherently high-risk if action in HIGH_RISK_ACTIONS and risk_score >= RISK_LEVELS["low"]: return True return False async def check_decision_guard( db: AsyncSession, tenant_id: uuid.UUID, instance_id: uuid.UUID, step_config: dict[str, Any], action: str, ai_use_case_metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: """Check the automated-decision guard for a workflow step. Returns a dict with: - ``allowed``: bool — whether the action can proceed automatically - ``requires_approval``: bool — whether human approval is needed - ``reason``: str — explanation if blocked - ``approval_action``: str — action to create approval for If the guard blocks, the workflow engine should pause and create an ApprovalRequest instead of executing the action. """ # Merge step-level and agent-level metadata step_metadata = step_config.get("ai_use_case_metadata") metadata = ai_use_case_metadata or step_metadata if not requires_human_review(metadata, action): return { "allowed": True, "requires_approval": False, "reason": None, } # Human review required risk_level = (metadata or {}).get("risk_level", "medium") return { "allowed": False, "requires_approval": True, "reason": ( f"Action '{action}' requires human review " f"(risk level: {risk_level})" ), "approval_action": action, "risk_level": risk_level, } __all__ = [ "requires_human_review", "check_decision_guard", "RISK_LEVELS", "HIGH_RISK_ACTIONS", ]