Traditional Model:
┌──────────────┐  ┌──────────────┐  ┌──────────────┐
│  CRUD Service │  │  Push Notify │  │ Webhook Svc  │
└──────┬───────┘  └──────┬───────┘  └──────┬───────┘
       │                 │                 │
       ▼                 ▼                 ▼
  Own transactions    Own retries       Own logging
  Own permissions     Own monitoring    Own formats

                 ┌───────────────────────────────────┐
                 │         ActionEngine               │
                 │                                   │
                 │  ┌─────────┐   ┌──────────────┐   │
                 │  │Scheduler│──→│ExecutorRouter │   │
                 │  └─────────┘   └──────┬───────┘   │
                 │                       │           │
                 │       ┌───────────────┼───────────┼───────┐
                 │       ▼               ▼           ▼       │
                 │  ┌─────────┐   ┌──────────┐  ┌────────┐  │
                 │  │CRUD Exec│   │Func Exec │  │Hook Exe│  │
                 │  └─────────┘   └──────────┘  └────────┘  │
                 │                                   │
                 │  ┌──────────────────────────────┐ │
                 │  │   Unified Audit Trail         │ │
                 │  └──────────────────────────────┘ │
                 └───────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                      ActionEngine                           │
│                                                             │
│  ┌──────────┐    ┌────────────┐    ┌──────────────────┐     │
│  │ Action   │───→│ Validation │───→│ Permission Check │     │
│  │ Request  │    │   Layer    │    │    (ABAC/RBAC)   │     │
│  └──────────┘    └────────────┘    └────────┬─────────┘     │
│                                             │               │
│                                             ▼               │
│  ┌──────────────────────────────────────────────────┐       │
│  │              ExecutorRouter                       │       │
│  │                                                  │       │
│  │  executor_type ──→ resolve_executor()            │       │
│  │                        │                         │       │
│  │    ┌───────────────────┼───────────────────┐     │       │
│  │    ▼         ▼         ▼         ▼         ▼     │       │
│  │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐   │       │
│  │ │Create│ │Update│ │Delete│ │Invoke│ │Compo-│   │       │
│  │ │Object│ │Object│ │Object│ │ Func │ │ site │   │       │
│  │ └──────┘ └──────┘ └──────┘ └──────┘ └──────┘   │       │
│  │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐   │       │
│  │ │Create│ │Delete│ │ Web- │ │Noti- │ │Simple│   │       │
│  │ │Relat.│ │Relat.│ │ hook │ │ficat.│ │  Op  │   │       │
│  │ └──────┘ └──────┘ └──────┘ └──────┘ └──────┘   │       │
│  └──────────────────────────────────────────────────┘       │
│                         │                                   │
│                         ▼                                   │
│  ┌──────────────────────────────────────────────────┐       │
│  │              Result Collector                     │       │
│  │  ┌────────┐  ┌─────────┐  ┌────────────────┐    │       │
│  │  │ Audit  │  │ Metrics │  │ Event Emission │    │       │
│  │  │  Log   │  │ Report  │  │ (CDC Stream)   │    │       │
│  │  └────────┘  └─────────┘  └────────────────┘    │       │
│  └──────────────────────────────────────────────────┘       │
└─────────────────────────────────────────────────────────────┘

from enum import Enum
from pydantic import BaseModel, Field
from typing import Any
from datetime import datetime
import uuid


class ExecutorType(str, Enum):
    """10 executor types"""
    CREATE_OBJECT = "CreateObject"
    UPDATE_OBJECT = "UpdateObject"
    DELETE_OBJECT = "DeleteObject"
    CREATE_RELATION = "CreateRelation"
    DELETE_RELATION = "DeleteRelation"
    INVOKE_FUNCTION = "InvokeFunction"
    WEBHOOK = "Webhook"
    NOTIFICATION = "Notification"
    SIMPLE_OP = "SimpleOp"
    COMPOSITE_OP = "CompositeOp"


class ActionStatus(str, Enum):
    PENDING = "pending"
    VALIDATING = "validating"
    EXECUTING = "executing"
    COMPENSATING = "compensating"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    ROLLED_BACK = "rolled_back"


class ActionRequest(BaseModel):
    """Unified Action request model"""
    action_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    executor_type: ExecutorType
    target_object_type: str | None = None
    target_object_id: str | None = None
    parameters: dict[str, Any] = Field(default_factory=dict)
    idempotency_key: str | None = None
    triggered_by: str = "system"
    context: dict[str, Any] = Field(default_factory=dict)
    created_at: datetime = Field(default_factory=datetime.utcnow)


class ActionResult(BaseModel):
    """Unified Action result model"""
    action_id: str
    status: ActionStatus
    executor_type: ExecutorType
    result_data: dict[str, Any] = Field(default_factory=dict)
    error_message: str | None = None
    started_at: datetime | None = None
    completed_at: datetime | None = None
    duration_ms: int | None = None
    audit_trail: list[dict[str, Any]] = Field(default_factory=list)

from abc import ABC, abstractmethod


class BaseExecutor(ABC):
    """Base class for all executors"""

    @property
    @abstractmethod
    def executor_type(self) -> ExecutorType:
        """Return the executor type identifier"""
        ...

    @abstractmethod
    async def validate(self, request: ActionRequest) -> list[str]:
        """
        Validate request parameters.
        Returns a list of errors; empty list means pass.
        """
        ...

    @abstractmethod
    async def execute(self, request: ActionRequest) -> ActionResult:
        """Execute the action and return the result"""
        ...

    @abstractmethod
    async def compensate(self, request: ActionRequest,
                         result: ActionResult) -> ActionResult:
        """
        Compensation operation for Saga rollback.
        Executes the inverse operation based on prior results.
        """
        ...

    def supports_dry_run(self) -> bool:
        """Whether dry-run mode is supported"""
        return False

    async def dry_run(self, request: ActionRequest) -> ActionResult:
        """Simulate execution without modifying data"""
        raise NotImplementedError("Dry-run not supported")

class CreateObjectExecutor(BaseExecutor):
    """Create an Ontology object"""

    @property
    def executor_type(self) -> ExecutorType:
        return ExecutorType.CREATE_OBJECT

    async def validate(self, request: ActionRequest) -> list[str]:
        errors: list[str] = []
        if not request.target_object_type:
            errors.append("target_object_type is required")
        params = request.parameters
        if "properties" not in params:
            errors.append("properties dict is required")
        # Validate against ObjectType schema
        schema = await self.ontology.get_object_type_schema(
            request.target_object_type
        )
        for field in schema.required_fields:
            if field not in params.get("properties", {}):
                errors.append(f"Missing required field: {field}")
        return errors

    async def execute(self, request: ActionRequest) -> ActionResult:
        started_at = datetime.utcnow()
        try:
            obj = await self.ontology.create_object(
                object_type=request.target_object_type,
                properties=request.parameters["properties"],
                created_by=request.triggered_by,
            )
            return ActionResult(
                action_id=request.action_id,
                status=ActionStatus.SUCCEEDED,
                executor_type=self.executor_type,
                result_data={"object_id": obj.id, "rid": obj.rid},
                started_at=started_at,
                completed_at=datetime.utcnow(),
            )
        except Exception as e:
            return ActionResult(
                action_id=request.action_id,
                status=ActionStatus.FAILED,
                executor_type=self.executor_type,
                error_message=str(e),
                started_at=started_at,
                completed_at=datetime.utcnow(),
            )

    async def compensate(self, request: ActionRequest,
                         result: ActionResult) -> ActionResult:
        """Compensate: delete the created object"""
        if result.status == ActionStatus.SUCCEEDED:
            object_id = result.result_data["object_id"]
            await self.ontology.delete_object(
                object_type=request.target_object_type,
                object_id=object_id,
                deleted_by="saga_compensator",
            )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.ROLLED_BACK,
            executor_type=self.executor_type,
        )

class CreateRelationExecutor(BaseExecutor):
    """Create a relation between Ontology objects"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        params = request.parameters
        relation = await self.ontology.create_relation(
            relation_type=params["relation_type"],
            source_object=params["source_rid"],
            target_object=params["target_rid"],
            properties=params.get("properties", {}),
        )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.SUCCEEDED,
            executor_type=self.executor_type,
            result_data={
                "relation_id": relation.id,
                "source_rid": params["source_rid"],
                "target_rid": params["target_rid"],
            },
        )

    async def compensate(self, request: ActionRequest,
                         result: ActionResult) -> ActionResult:
        """Compensate: delete the created relation"""
        await self.ontology.delete_relation(
            relation_id=result.result_data["relation_id"]
        )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.ROLLED_BACK,
            executor_type=self.executor_type,
        )

class InvokeFunctionExecutor(BaseExecutor):
    """Invoke a user-defined function"""

    def __init__(self, function_runtime, timeout_seconds: int = 30):
        self.runtime = function_runtime
        self.timeout = timeout_seconds

    async def execute(self, request: ActionRequest) -> ActionResult:
        params = request.parameters
        function_ref = params["function_rid"]
        func_args = params.get("arguments", {})

        # Dispatch to the appropriate sandbox via FunctionRuntime
        execution = await self.runtime.invoke(
            function_rid=function_ref,
            arguments=func_args,
            timeout_seconds=self.timeout,
            caller_context=request.context,
        )

        return ActionResult(
            action_id=request.action_id,
            status=(ActionStatus.SUCCEEDED
                    if execution.success
                    else ActionStatus.FAILED),
            executor_type=self.executor_type,
            result_data={
                "return_value": execution.return_value,
                "execution_id": execution.id,
                "runtime_ms": execution.runtime_ms,
            },
            error_message=execution.error if not execution.success else None,
        )

    async def compensate(self, request: ActionRequest,
                         result: ActionResult) -> ActionResult:
        """Invoke the corresponding compensation function if defined"""
        params = request.parameters
        compensate_ref = params.get("compensate_function_rid")
        if compensate_ref:
            await self.runtime.invoke(
                function_rid=compensate_ref,
                arguments={
                    "original_result": result.result_data,
                    "original_args": params.get("arguments", {}),
                },
            )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.ROLLED_BACK,
            executor_type=self.executor_type,
        )

class WebhookExecutor(BaseExecutor):
    """Invoke an external webhook"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        params = request.parameters
        url = params["url"]
        method = params.get("method", "POST")
        headers = params.get("headers", {})
        payload = params.get("payload", {})
        retry_config = params.get("retry", {
            "max_retries": 3, "backoff": "exponential"
        })

        # Signature verification
        if "signing_secret" in params:
            headers["X-Onto-Signature"] = self._compute_signature(
                payload, params["signing_secret"]
            )

        response = await self.http_client.request(
            method=method,
            url=url,
            json=payload,
            headers=headers,
            timeout=params.get("timeout_seconds", 30),
            retry_config=retry_config,
        )

        return ActionResult(
            action_id=request.action_id,
            status=(ActionStatus.SUCCEEDED
                    if 200 <= response.status_code < 300
                    else ActionStatus.FAILED),
            executor_type=self.executor_type,
            result_data={
                "status_code": response.status_code,
                "response_body": response.json(),
            },
        )

class NotificationExecutor(BaseExecutor):
    """Send notifications across 9 channels"""

    SUPPORTED_CHANNELS = [
        "email", "sms", "webhook", "slack", "dingtalk",
        "wechat_work", "feishu", "in_app", "push",
    ]

    async def execute(self, request: ActionRequest) -> ActionResult:
        params = request.parameters
        channel = params["channel"]
        template_id = params.get("template_id")
        recipients = params["recipients"]

        # Render template
        content = await self.template_engine.render(
            template_id=template_id,
            variables=params.get("variables", {}),
        )

        # Dispatch to channel adapter
        adapter = self.channel_registry.get(channel)
        delivery_results = await adapter.send(
            recipients=recipients,
            content=content,
            metadata=params.get("metadata", {}),
        )

        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.SUCCEEDED,
            executor_type=self.executor_type,
            result_data={
                "channel": channel,
                "delivered": len([r for r in delivery_results if r.success]),
                "failed": len([r for r in delivery_results if not r.success]),
            },
        )

class SimpleOpExecutor(BaseExecutor):
    """Execute simple inline operations"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        op_type = request.parameters["op"]
        match op_type:
            case "set_variable":
                key = request.parameters["key"]
                value = request.parameters["value"]
                request.context[key] = value
            case "log":
                level = request.parameters.get("level", "info")
                message = request.parameters["message"]
                self.logger.log(level, message)
            case "delay":
                seconds = request.parameters["seconds"]
                await asyncio.sleep(seconds)
            case "assert":
                condition = request.parameters["condition"]
                if not self._evaluate_condition(condition, request.context):
                    return ActionResult(
                        action_id=request.action_id,
                        status=ActionStatus.FAILED,
                        executor_type=self.executor_type,
                        error_message=f"Assertion failed: {condition}",
                    )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.SUCCEEDED,
            executor_type=self.executor_type,
        )

class CompositeOpExecutor(BaseExecutor):
    """Orchestrate multiple sub-Actions"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        steps = request.parameters["steps"]
        mode = request.parameters.get("mode", "sequential")
        # mode: sequential | parallel | conditional

        results: list[ActionResult] = []

        if mode == "sequential":
            for step in steps:
                sub_request = ActionRequest(**step)
                sub_result = await self.engine.dispatch(sub_request)
                results.append(sub_result)
                if sub_result.status == ActionStatus.FAILED:
                    if request.parameters.get("stop_on_failure", True):
                        break

        elif mode == "parallel":
            tasks = [
                self.engine.dispatch(ActionRequest(**step))
                for step in steps
            ]
            results = await asyncio.gather(*tasks, return_exceptions=True)

        elif mode == "conditional":
            for step in steps:
                condition = step.get("when")
                if condition and not self._evaluate(condition, request.context):
                    continue
                sub_request = ActionRequest(**step)
                sub_result = await self.engine.dispatch(sub_request)
                results.append(sub_result)

        all_succeeded = all(
            r.status == ActionStatus.SUCCEEDED for r in results
        )
        return ActionResult(
            action_id=request.action_id,
            status=(ActionStatus.SUCCEEDED
                    if all_succeeded else ActionStatus.FAILED),
            executor_type=self.executor_type,
            result_data={
                "mode": mode,
                "total_steps": len(steps),
                "completed": len([
                    r for r in results
                    if r.status == ActionStatus.SUCCEEDED
                ]),
                "sub_results": [r.model_dump() for r in results],
            },
        )

┌──────────┐     ┌────────────┐     ┌───────────────┐     ┌───────────┐
│ PENDING  │────→│ VALIDATING │────→│  EXECUTING    │────→│ SUCCEEDED │
└──────────┘     └─────┬──────┘     └──────┬────────┘     └───────────┘
                       │                   │
                       │ validation        │ execution
                       │ failed            │ failed
                       ▼                   ▼
                 ┌──────────┐       ┌──────────────┐     ┌───────────┐
                 │  FAILED  │       │ COMPENSATING │────→│ROLLED_BACK│
                 └──────────┘       └──────────────┘     └───────────┘

class ActionScheduler:
    """Unified Action scheduler"""

    def __init__(self, executor_registry: dict[ExecutorType, BaseExecutor]):
        self.executors = executor_registry
        self.audit_log = AuditLogger()
        self.metrics = MetricsCollector()
        self.idempotency_store = IdempotencyStore()

    async def dispatch(self, request: ActionRequest) -> ActionResult:
        """Unified dispatch entry point"""
        # 1. Idempotency check
        if request.idempotency_key:
            cached = await self.idempotency_store.get(
                request.idempotency_key
            )
            if cached:
                return cached

        # 2. Resolve executor
        executor = self.executors.get(request.executor_type)
        if not executor:
            raise ValueError(
                f"Unknown executor: {request.executor_type}"
            )

        # 3. Permission check
        await self._check_permissions(request)

        # 4. Validate parameters
        errors = await executor.validate(request)
        if errors:
            result = ActionResult(
                action_id=request.action_id,
                status=ActionStatus.FAILED,
                executor_type=request.executor_type,
                error_message=f"Validation failed: {'; '.join(errors)}",
            )
            await self.audit_log.record(
                request, result, phase="validation"
            )
            return result

        # 5. Execute
        await self.audit_log.record(request, phase="start")
        self.metrics.increment("action.dispatched", tags={
            "executor_type": request.executor_type.value
        })

        with self.metrics.timer("action.execution_time", tags={
            "executor_type": request.executor_type.value
        }):
            result = await executor.execute(request)

        # 6. Record audit log
        await self.audit_log.record(request, result, phase="complete")

        # 7. Cache idempotency result
        if request.idempotency_key:
            await self.idempotency_store.set(
                request.idempotency_key, result, ttl=3600
            )

        # 8. Emit CDC event
        await self._emit_cdc_event(request, result)

        return result

    async def _check_permissions(self, request: ActionRequest) -> None:
        """ABAC permission check"""
        decision = await self.policy_engine.evaluate(
            subject=request.triggered_by,
            action=request.executor_type.value,
            resource=(
                f"{request.target_object_type}"
                f"/{request.target_object_id}"
            ),
            context=request.context,
        )
        if not decision.allowed:
            raise PermissionError(
                f"Action denied: {decision.reason}"
            )

    async def _emit_cdc_event(self, request: ActionRequest,
                               result: ActionResult) -> None:
        """Emit a Change Data Capture event"""
        if result.status == ActionStatus.SUCCEEDED:
            event = {
                "type": "action.completed",
                "action_id": request.action_id,
                "executor_type": request.executor_type.value,
                "target": {
                    "object_type": request.target_object_type,
                    "object_id": request.target_object_id,
                },
                "result": result.result_data,
                "timestamp": result.completed_at.isoformat(),
            }
            await self.event_bus.publish("ontology.actions", event)

from typing import Type


class ExecutorRegistry:
    """Executor registry"""

    _executors: dict[ExecutorType, BaseExecutor] = {}

    @classmethod
    def register(cls, executor_cls: Type[BaseExecutor]) -> Type[BaseExecutor]:
        """Decorator: auto-register an executor"""
        instance = executor_cls()
        cls._executors[instance.executor_type] = instance
        return executor_cls

    @classmethod
    def get(cls, executor_type: ExecutorType) -> BaseExecutor:
        executor = cls._executors.get(executor_type)
        if not executor:
            raise KeyError(
                f"No executor registered for {executor_type}"
            )
        return executor

    @classmethod
    def list_all(cls) -> dict[ExecutorType, BaseExecutor]:
        return dict(cls._executors)


# Using the decorator
@ExecutorRegistry.register
class CreateObjectExecutor(BaseExecutor):
    ...

@ExecutorRegistry.register
class UpdateObjectExecutor(BaseExecutor):
    ...

class TransactionalExecutor(BaseExecutor):
    """Transactional wrapper around an executor"""

    def __init__(self, inner: BaseExecutor, db_session_factory):
        self.inner = inner
        self.session_factory = db_session_factory

    async def execute(self, request: ActionRequest) -> ActionResult:
        async with self.session_factory() as session:
            try:
                result = await self.inner.execute(request)
                if result.status == ActionStatus.SUCCEEDED:
                    await session.commit()
                else:
                    await session.rollback()
                return result
            except Exception:
                await session.rollback()
                raise

class SagaCompositeExecutor(CompositeOpExecutor):
    """Saga-based composite executor"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        steps = request.parameters["steps"]
        completed: list[tuple[ActionRequest, ActionResult]] = []

        try:
            for step_data in steps:
                sub_request = ActionRequest(**step_data)
                sub_result = await self.engine.dispatch(sub_request)
                completed.append((sub_request, sub_result))

                if sub_result.status == ActionStatus.FAILED:
                    await self._compensate_all(completed)
                    return ActionResult(
                        action_id=request.action_id,
                        status=ActionStatus.ROLLED_BACK,
                        executor_type=self.executor_type,
                        error_message=(
                            f"Step failed: {sub_result.error_message}"
                        ),
                    )

            return ActionResult(
                action_id=request.action_id,
                status=ActionStatus.SUCCEEDED,
                executor_type=self.executor_type,
            )

        except Exception as e:
            await self._compensate_all(completed)
            raise

    async def _compensate_all(
        self, completed: list[tuple[ActionRequest, ActionResult]]
    ) -> None:
        """Compensate all completed steps in reverse order"""
        for req, result in reversed(completed):
            executor = self.engine.executors.get(req.executor_type)
            if executor:
                try:
                    await executor.compensate(req, result)
                except Exception as e:
                    self.logger.error(
                        f"Compensation failed for {req.action_id}: {e}"
                    )
                    await self.dead_letter_queue.enqueue(req, result, e)

METRICS = {
    "action_dispatched_total": Counter(
        "action_dispatched_total",
        "Total number of actions dispatched",
        labels=["executor_type", "status"],
    ),
    "action_execution_duration_seconds": Histogram(
        "action_execution_duration_seconds",
        "Action execution duration",
        labels=["executor_type"],
        buckets=[0.01, 0.05, 0.1, 0.5, 1, 5, 10, 30, 60],
    ),
    "action_compensation_total": Counter(
        "action_compensation_total",
        "Total compensations triggered",
        labels=["executor_type", "reason"],
    ),
    "action_dead_letter_total": Counter(
        "action_dead_letter_total",
        "Actions sent to dead letter queue",
        labels=["executor_type"],
    ),
}

┌────────────┬──────────────────────────────────────────────────────┐
│ Timestamp  │ Audit Record                                         │
├────────────┼──────────────────────────────────────────────────────┤
│ T+0ms      │ [DISPATCH] action_id=abc executor=CreateObject       │
│ T+2ms      │ [VALIDATE] passed, 0 errors                         │
│ T+5ms      │ [PERMISSION] allowed by policy: ontology.write       │
│ T+8ms      │ [EXECUTE]  started                                   │
│ T+45ms     │ [EXECUTE]  completed, object_id=obj-123              │
│ T+48ms     │ [CDC]      event emitted to ontology.actions          │
│ T+50ms     │ [AUDIT]    full record persisted                      │
└────────────┴──────────────────────────────────────────────────────┘

class ExecutorPool:
    """Maintain a connection pool for high-frequency executors"""

    def __init__(self, max_connections: int = 100):
        self.pool = asyncio.Semaphore(max_connections)
        self.active_count = 0

    async def execute_with_pool(
        self, executor: BaseExecutor, request: ActionRequest
    ) -> ActionResult:
        async with self.pool:
            self.active_count += 1
            try:
                return await executor.execute(request)
            finally:
                self.active_count -= 1