"""Automation Rule Executor. Entry points ------------ ``execute_rule(rule, trigger_payload)`` Synchronous execution — evaluates conditions, runs actions in order, persists an ``AutomationExecution`` record. Called directly from tests and the Celery task. ``execute_rule_async(rule_id, trigger_payload)`` Thin shim that dispatches to the Celery task ``execute_automation_rule_task``. Import-safe: Celery is optional at test time. Retry contract (task 3.5.4) ---------------------------- The Celery task is configured with ``max_retries=3``, exponential countdown. The executor records per-action ``{"action": key, "status": "ok"|"error", "error": msg}`` in ``AutomationExecution.actions_executed``. On a retry, the executor skips actions that already succeeded in a previous attempt (checked via the ``already_ok`` set built from the existing execution row). """ from __future__ import annotations import uuid from typing import Any import structlog from django.db import IntegrityError, transaction from django.utils import timezone _log = structlog.get_logger("simorgh.automation.executor") class ExecutionError(RuntimeError): """Raised when a rule execution fails after all retries.""" # --------------------------------------------------------------------------- # Stats helper # --------------------------------------------------------------------------- def _update_rule_stats(rule_id: int, status: str) -> None: """Bump run_count, last_run_at, last_status atomically.""" from django.db.models import F from simorgh.apps.automation.models import AutomationRule AutomationRule.objects.filter(pk=rule_id).update( run_count=F("run_count") + 1, last_run_at=timezone.now(), last_status=status, ) # --------------------------------------------------------------------------- # Core synchronous executor # --------------------------------------------------------------------------- def execute_rule( rule: Any, # AutomationRule — typed as Any to avoid import at module level trigger_payload: dict[str, Any], *, actor_id: int | None = None, trigger_event: str = "", idempotency_key: str = "", ) -> Any: # → AutomationExecution """Execute *rule* synchronously against *trigger_payload*. Steps ----- 1. Derive a stable *idempotency_key* (callers may supply one; otherwise a fresh UUID is generated — useful for manual/scheduled triggers). 2. Attempt to get-or-create an ``AutomationExecution`` row. If the row already exists and is not RUNNING (e.g. a completed retry), return it immediately — idempotent. 3. Evaluate ``rule.conditions`` against *trigger_payload*. Mismatch → mark SKIPPED and return. 4. Execute each action in ``rule.actions`` in order. Already-succeeded actions (from a previous partial run) are skipped. 5. Persist final status + action log; update rule stats. Returns ------- ``AutomationExecution`` instance with the final status. """ from simorgh.apps.automation.evaluator import evaluate_conditions from simorgh.apps.automation.models import AutomationExecution, ExecutionStatus from simorgh.apps.automation.registry import ActionContext, get_action if not idempotency_key: idempotency_key = str(uuid.uuid4()) # ------------------------------------------------------------------ # 2. Get-or-create execution row # ------------------------------------------------------------------ try: with transaction.atomic(): execution, created = AutomationExecution.objects.get_or_create( rule=rule, idempotency_key=idempotency_key, defaults={ "tenant": rule.tenant, "organization_node": rule.organization_node, "rule_version": rule.version, "trigger_event": trigger_event, "trigger_payload": trigger_payload, "status": ExecutionStatus.RUNNING, "started_at": timezone.now(), }, ) except IntegrityError: # Race: another worker already created the row — load it. execution = AutomationExecution.objects.get( rule=rule, idempotency_key=idempotency_key ) created = False # Already finished by a concurrent worker — return as-is. if not created and execution.status != ExecutionStatus.RUNNING: return execution # ------------------------------------------------------------------ # 3. Evaluate conditions # ------------------------------------------------------------------ try: matched = evaluate_conditions(rule.conditions, trigger_payload) except Exception as exc: execution.status = ExecutionStatus.FAILED execution.error_message = f"condition evaluation error: {exc}" execution.finished_at = timezone.now() execution.save(update_fields=["status", "error_message", "finished_at"]) _update_rule_stats(rule.pk, ExecutionStatus.FAILED) return execution if not matched: execution.status = ExecutionStatus.SKIPPED execution.finished_at = timezone.now() execution.save(update_fields=["status", "finished_at"]) _update_rule_stats(rule.pk, ExecutionStatus.SKIPPED) _log.debug( "automation.rule.skipped", rule_id=rule.pk, idempotency_key=idempotency_key, ) return execution # ------------------------------------------------------------------ # 4. Execute actions # ------------------------------------------------------------------ # Build the set of actions that already succeeded in a previous attempt # so we can skip them safely (idempotent retry). already_ok: set[str] = { entry["action"] for entry in execution.actions_executed if entry.get("status") == "ok" } actions_log: list[dict[str, Any]] = list(execution.actions_executed) final_status = ExecutionStatus.SUCCESS for action_def in rule.actions: action_key: str = action_def.get("action", "") params: dict[str, Any] = action_def.get("params", {}) # Skip actions that already ran OK in a previous (partial) attempt. if action_key in already_ok: _log.debug( "automation.action.skipped_on_retry", action=action_key, rule_id=rule.pk, ) continue try: spec = get_action(action_key) except KeyError: entry = { "action": action_key, "status": "error", "result": None, "error": f"Unknown action key: {action_key!r}", } actions_log.append(entry) final_status = ExecutionStatus.FAILED # Abort: unknown action is a config error, not transient. break ctx = ActionContext( tenant_id=rule.tenant_id, rule_id=rule.pk, trigger_event=trigger_event, trigger_payload=trigger_payload, params=params, actor_id=actor_id, ) try: spec.handler(ctx) actions_log.append({ "action": action_key, "status": "ok", "result": None, "error": None, }) _log.info( "automation.action.ok", action=action_key, rule_id=rule.pk, ) except Exception as exc: actions_log.append({ "action": action_key, "status": "error", "result": None, "error": str(exc), }) final_status = ExecutionStatus.FAILED _log.warning( "automation.action.failed", action=action_key, rule_id=rule.pk, error=str(exc), ) # Continue to remaining actions (resilient mode). # Transient errors will surface on Celery retry. # ------------------------------------------------------------------ # 5. Persist final state # ------------------------------------------------------------------ execution.actions_executed = actions_log execution.status = final_status execution.finished_at = timezone.now() execution.save(update_fields=["actions_executed", "status", "finished_at"]) _update_rule_stats(rule.pk, final_status) _log.info( "automation.rule.finished", rule_id=rule.pk, status=final_status, idempotency_key=idempotency_key, ) return execution # --------------------------------------------------------------------------- # Async shim # --------------------------------------------------------------------------- def execute_rule_async(rule_id: int, trigger_payload: dict[str, Any], **kwargs: Any) -> None: """Dispatch ``execute_rule`` to a Celery worker. Imported lazily so the module is usable without Celery in unit tests. """ from simorgh.apps.automation.tasks import execute_automation_rule_task # noqa: PLC0415 execute_automation_rule_task.delay(rule_id, trigger_payload, **kwargs)