orchestrata_ultimate.governance.event_emitting_policy_engine

Reference › orchestrata_ultimate › governance › event_emitting_policy_engine

A decorator around PolicyEngine that publishes governance events, so the gateway and observability layers can follow decisions without being coupled to the engine. The package initialiser re-exports it. EventEmittingPolicyEngine takes the wrapped engine and an optional event bus (IEventBus); __getattr__ forwards every other attribute to the engine. Its evaluate calls the engine, then emits a governance.decision.made event carrying the action, actor, allowed flag, reason, cache-hit flag and duration, and, when the action was not allowed, a second governance.violation.detected event. A failure to emit is logged and swallowed, and the decision is returned unchanged; an exception from the wrapped engine propagates. The events go to the kernel EventBus when that is the bus supplied.

At a glance

Class diagram

Dependency diagram

API

Classes

EventEmittingPolicyEngine

Line 13

PolicyEngine wrapper that emits system events on governance decisions.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/governance/event_emitting_policy_engine.py

"""Event-emitting wrapper for PolicyEngine governance decisions."""

import logging
from typing import Any

from orchestrata_ultimate.core.protocols import IEventBus, SystemEvent
from orchestrata_ultimate.governance.policy_engine import PolicyEngine
from orchestrata_ultimate.governance.types import PolicyDecision

logger = logging.getLogger(__name__)


class EventEmittingPolicyEngine:
    """PolicyEngine wrapper that emits system events on governance decisions.

    Wraps a PolicyEngine instance and emits governance events to an optional
    IEventBus after each evaluate() call. When event_bus is None the wrapper
    is a transparent pass-through.

    Events emitted:
        governance.decision.made — on every evaluate() call
        governance.violation.detected — additionally when decision.allowed is False
    """

    def __init__(
        self,
        policy_engine: PolicyEngine,
        event_bus: IEventBus | None = None,
    ) -> None:
        """Initialise the wrapper.

        Args:
            policy_engine: The PolicyEngine instance to wrap.
            event_bus: Optional event bus for emission. No events are emitted
                if None.
        """
        self._policy_engine = policy_engine
        self._event_bus = event_bus

    def __getattr__(self, name: str) -> Any:
        """Delegate all other attribute access to the wrapped PolicyEngine."""
        return getattr(self._policy_engine, name)

    async def evaluate(self, action: str, context: dict[str, Any]) -> PolicyDecision:
        """Evaluate an action and emit governance events.

        Args:
            action: The action to evaluate.
            context: Evaluation context.

        Returns:
            PolicyDecision from the wrapped engine.

        Raises:
            Any exception raised by the wrapped engine is re-raised after
            logging — event emission failures are swallowed.
        """
        decision: PolicyDecision = await self._policy_engine.evaluate(action, context)

        if self._event_bus is not None:
            try:
                await self._event_bus.emit(
                    SystemEvent(
                        type="governance.decision.made",
                        source="EventEmittingPolicyEngine",
                        payload={
                            "action": action,
                            "actor": decision.actor,
                            "allowed": decision.allowed,
                            "reason": decision.reason,
                            "cache_hit": decision.cache_hit,
                            "duration_ms": decision.duration_ms,
                        },
                    )
                )
                if not decision.allowed:
                    await self._event_bus.emit(
                        SystemEvent(
                            type="governance.violation.detected",
                            source="EventEmittingPolicyEngine",
                            payload={
                                "action": action,
                                "actor": decision.actor,
                                "reason": decision.reason,
                            },
                        )
                    )
            except Exception:
                logger.exception(
                    "Failed to emit governance events for action=%s", action
                )

        return decision