Decentralized Contract Architecture for Action Handlers

flexirule/ruleflow/core/action_handlers/__init__.py

@@ -25,6 +25,7 @@ def execute(self, action, context, engine):
 from typing import TYPE_CHECKING, Any, ClassVar, Optional
 
 if TYPE_CHECKING:
+	from flexirule.ruleflow.core.action_handlers.base_contract import ActionContract, OperationContract
 	from flexirule.ruleflow.core.engine import RuleEngine
 
 
@@ -38,6 +39,88 @@ class ActionHandler(ABC):
 
 	action_type: str | None = None  # Must be set by subclass
 
+	@classmethod
+	def get_action_contract(cls) -> "ActionContract":
+		"""Return the action type's top-level contract.
+
+		Override in subclass. Default returns a minimal contract.
+		"""
+		from flexirule.ruleflow.core.action_handlers.base_contract import ActionContract
+
+		return ActionContract(
+			action_type=cls.action_type or "",
+			required_fields=[],
+			has_next_true=True,
+			has_next_false=False,
+			terminal=False,
+		)
+
+	@classmethod
+	def get_operation_contracts(cls) -> dict[str, "OperationContract"]:
+		"""Return operation-level contracts keyed by operation name.
+
+		Override in handlers with operations (Query Records, Document Action, etc.).
+		Default returns empty dict (no operations).
+		"""
+		return {}
+
+	@classmethod
+	def get_runtime_policy(cls, operation=None, process_operation=None) -> dict:
+		"""Resolve runtime policy for this handler + optional operation.
+
+		Override for handlers with complex policy resolution (e.g., Process).
+		Default derives policy from the action contract + operation policies.
+		"""
+		contract_dict = cls.get_action_contract().to_dict()
+		policy: dict = {
+			"allowed_mutations": list(contract_dict.get("allowed_mutations", []) or []),
+			"allowed_return_types": list(contract_dict.get("allowed_return_types", []) or []),
+			"default_return_type": contract_dict.get("default_return_type"),
+			"field_labels": dict(contract_dict.get("field_labels", {}) or {}),
+			"show_return_type": contract_dict.get("show_return_type"),
+			"require_return_type": contract_dict.get("require_return_type", False),
+			"show_return_variable": contract_dict.get("show_return_variable"),
+			"require_return_variable": contract_dict.get("require_return_variable", False),
+		}
+
+		if operation:
+			op_policy = dict((contract_dict.get("operation_policies", {}) or {}).get(operation, {}) or {})
+
+			# Merge operation contract action_overrides into policy if relevant
+			op_contracts = cls.get_operation_contracts()
+			if operation in op_contracts:
+				op_contract = op_contracts[operation]
+				for field_def in op_contract.action_overrides:
+					fieldname = field_def.get("fieldname")
+					if fieldname == "mutation_mode" and field_def.get("options"):
+						op_policy["allowed_mutations"] = field_def["options"]
+					elif fieldname == "return_type":
+						if field_def.get("options"):
+							op_policy["allowed_return_types"] = field_def["options"]
+						if field_def.get("default"):
+							op_policy["default_return_type"] = field_def["default"]
+						if "read_only" in field_def:
+							op_policy["show_return_type"] = not field_def.get("read_only", False)
+							op_policy["require_return_type"] = field_def.get("reqd", False)
+
+			for key in ("allowed_mutations", "allowed_return_types"):
+				if op_policy.get(key):
+					policy[key] = list(op_policy[key])
+			if op_policy.get("default_return_type"):
+				policy["default_return_type"] = op_policy["default_return_type"]
+			if op_policy.get("show_return_type") is not None:
+				policy["show_return_type"] = op_policy.get("show_return_type")
+			if op_policy.get("require_return_type") is not None:
+				policy["require_return_type"] = op_policy.get("require_return_type")
+			if op_policy.get("show_return_variable") is not None:
+				policy["show_return_variable"] = op_policy.get("show_return_variable")
+			if op_policy.get("require_return_variable") is not None:
+				policy["require_return_variable"] = op_policy.get("require_return_variable")
+			if op_policy.get("field_labels"):
+				policy["field_labels"].update(op_policy.get("field_labels", {}))
+
+		return policy
+
 	@abstractmethod
 	def execute(self, action, context: dict, engine: "RuleEngine") -> tuple[Any, str | None]:
 		"""
@@ -331,6 +414,8 @@ class HandlerRegistry:
 	"""
 
 	_handlers: ClassVar[dict] = {}
+	_contract_cache: ClassVar[dict | None] = None
+	_op_contract_cache: ClassVar[dict | None] = None
 	_initialized: bool = False
 
 	@classmethod
@@ -347,6 +432,8 @@ def register(cls, handler: ActionHandler) -> None:
 		if not handler.action_type:
 			raise ValueError(f"Handler {handler.__class__.__name__} must set 'action_type' class attribute")
 		cls._handlers[handler.action_type] = handler
+		cls._contract_cache = None
+		cls._op_contract_cache = None
 
 	@classmethod
 	def get(cls, action_type: str) -> ActionHandler | None:
@@ -384,6 +471,115 @@ def action_types(cls) -> list:
 		cls._ensure_initialized()
 		return list(cls._handlers.keys())
 
+	@classmethod
+	def get_action_contract(cls, action_type: str) -> dict:
+		"""Get contract for a specific action type."""
+		from flexirule.ruleflow.core.contract_utils import normalize_action_type
+
+		cls._ensure_initialized()
+		normalized = normalize_action_type(action_type, cls.action_types())
+		handler = cls.get(normalized)
+		if handler:
+			return handler.get_action_contract().to_dict()
+		return {
+			"required_fields": [],
+			"has_next_true": True,
+			"has_next_false": False,
+			"terminal": False,
+			"css": {},
+			"node_type": "action",
+			"category": "Other",
+			"configurable": True,
+		}
+
+	@classmethod
+	def get_all_contracts(cls) -> dict[str, dict]:
+		"""Aggregate all action contracts. Cached for performance."""
+		if cls._contract_cache is not None:
+			return cls._contract_cache
+
+		cls._ensure_initialized()
+		cls._contract_cache = {at: h.get_action_contract().to_dict() for at, h in cls._handlers.items()}
+		return cls._contract_cache
+
+	@classmethod
+	def get_operation_contract(cls, operation: str) -> dict:
+		"""Get operation contract from the owning handler."""
+		cls._ensure_initialized()
+
+		# Check all handlers to find which one owns this operation
+		for handler in cls._handlers.values():
+			contracts = handler.get_operation_contracts()
+			if operation in contracts:
+				return contracts[operation].to_dict()
+
+		# Fallback: check if the operation IS an action type
+		from flexirule.ruleflow.core.contract_utils import normalize_action_type
+
+		normalized_at = normalize_action_type(operation, cls.action_types())
+		if normalized_at in cls._handlers:
+			handler = cls._handlers[normalized_at]
+			base_contract = handler.get_action_contract().to_dict()
+			return {
+				"Rule": [],
+				"Rule Action": [
+					{"fieldname": "action_type", "default": normalized_at},
+					{"fieldname": "operation", "default": operation if operation != normalized_at else None},
+					{
+						"fieldname": "description",
+						"description": base_contract.get("description", f"Executes {operation}"),
+					},
+				]
+				+ [{"fieldname": f, "reqd": 1} for f in base_contract.get("required_fields", [])],
+				"Validation": base_contract.get("validation", {}),
+			}
+
+		return {
+			"Rule": [],
+			"Rule Action": [{"fieldname": "description", "description": f"Unknown operation: {operation}"}],
+			"Validation": {},
+		}
+
+	@classmethod
+	def get_all_operation_contracts(cls) -> dict[str, dict]:
+		"""Aggregate all operation contracts from all handlers."""
+		if cls._op_contract_cache is not None:
+			return cls._op_contract_cache
+
+		cls._ensure_initialized()
+		result = {}
+
+		# Add basic operation contracts for handlers without specific operations
+		for at, handler in cls._handlers.items():
+			# If handler has explicit operations, they will be added below
+			ops = handler.get_operation_contracts()
+			if not ops:
+				# Add a default operation contract for the action type itself
+				result[at] = cls.get_operation_contract(at)
+			else:
+				for op_name, oc in ops.items():
+					result[op_name] = oc.to_dict()
+
+		cls._op_contract_cache = result
+		return result
+
+	@classmethod
+	def get_effective_policy(
+		cls,
+		action_type: str | None,
+		operation: str | None = None,
+		process_operation: dict | None = None,
+	) -> dict:
+		"""Delegate policy resolution to the handler."""
+		from flexirule.ruleflow.core.contract_utils import normalize_action_type
+
+		cls._ensure_initialized()
+		normalized_at = normalize_action_type(action_type, cls.action_types())
+		handler = cls.get(normalized_at)
+		if handler:
+			return handler.get_runtime_policy(operation, process_operation)
+		return {}
+
 	@classmethod
 	def _ensure_initialized(cls) -> None:
 		"""

flexirule/ruleflow/core/action_handlers/assignment.py

@@ -9,6 +9,10 @@
 from frappe import _
 
 from flexirule.ruleflow.core.action_handlers import ActionHandler, HandlerRegistry
+from flexirule.ruleflow.core.action_handlers.base_contract import (
+	ActionContract,
+	OperationContract,
+)
 from flexirule.ruleflow.core.action_plan_cache import get_action_plan
 from flexirule.ruleflow.core.context_manager import ContextManager
 from flexirule.ruleflow.core.exceptions import MethodExecutionError
@@ -21,6 +25,41 @@ class AssignmentHandler(ActionHandler):
 
 	action_type = "Assignment"
 
+	@classmethod
+	def get_action_contract(cls):
+		return ActionContract(
+			action_type="Assignment",
+			required_fields=["config"],
+			has_next_true=True,
+			has_next_false=False,
+			terminal=False,
+			css={"icon": "fa fa-list-ol", "color": "#14b8a6"},
+			field_labels={
+				"config": "Assignments",
+			},
+			node_type="assignment",
+			category="Data Actions",
+			configurable=True,
+			config_component="AssignmentConfig",
+		)
+
+	@classmethod
+	def get_operation_contracts(cls):
+		return {
+			"Assignment": OperationContract(
+				operation="Assignment",
+				action_overrides=[
+					{"fieldname": "action_type", "default": "Assignment"},
+					{"fieldname": "config", "reqd": 1, "description": "Array of assignments (JSON)"},
+					{
+						"fieldname": "description",
+						"description": "Updates document fields or context variables in batch",
+					},
+				],
+				validation={"backend": "validate_assignment"},
+			)
+		}
+
 	def execute(self, action, context, engine):
 		"""
 		Execute Assignment action - sequential batch mutations.