orchestrata_ultimate.gateway.handlers.governance

Reference › orchestrata_ultimate › gateway › handlers › governance

GovernanceHandler exposes the constitutional governance engine over RPC: governance.decisions, governance.query, governance.evaluate and governance.rules. rpc creates it and server supplies the policy engine or its factory, the graph repository and the message processor through setters. See the governance package and gateway.

• handle_evaluate is a dry-run check. With a policy engine wired it evaluates the intent with a context that carries the actor, publishes a decision event on the governance topics through publish_decision_event, and returns allowed or denied with the reason, decision type and articles checked. An evaluation error is treated as a denial. With no engine it returns a permissive stub result that says so.
• handle_query reads recent PolicyDecision nodes from the graph audit trail, newest first, with an optional status filter and a limit capped at 1000, under a timeout. It returns an empty list with an error text on failure.
• handle_rules lists the constitutional articles (number, title, description, priority, predicates) when an engine is configured, otherwise an empty list.
• handle_decisions subscribes the connection to the governance.decisions topic, with optional filters, and _resolve_engine builds the engine lazily from the factory.

At a glance

Class diagram

Dependency diagram

API

Classes

GovernanceHandler

Line 19

Governance decision streaming and query methods.

Source

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

"""Governance RPC handlers -- decisions, query, evaluate, rules.

Implements SPEC-014 FR-018 governance methods.
"""

import asyncio
import json
import logging
import uuid
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.pubsub import PubSubManager

logger = logging.getLogger(__name__)


class GovernanceHandler:
    """Governance decision streaming and query methods."""

    def __init__(self, pubsub: PubSubManager) -> None:
        """Wire handler to governance subsystems.

        Args:
            pubsub: Pub/sub subscription manager.
        """
        self._pubsub = pubsub
        self._graph_repository: Any = None
        self._policy_engine: Any = None
        self._engine_factory: Any = None
        self._processor: Any = None

    def set_processor(self, processor: Any) -> None:
        """Inject the message processor after construction.

        Args:
            processor: MessageProcessor instance.
        """
        self._processor = processor

    def set_graph_repository(self, graph_repository: Any) -> None:
        """Inject the operational-graph repository for historical decision
        queries.

        Args:
            graph_repository: An AsyncGraphRepository instance.
        """
        self._graph_repository = graph_repository

    def set_policy_engine(self, engine: Any) -> None:
        """Inject a pre-built governance PolicyEngine for dry-run evaluation.

        Args:
            engine: A PolicyEngine (or EventEmittingPolicyEngine) exposing
                ``async evaluate(action, context) -> PolicyDecision``.
        """
        self._policy_engine = engine

    def set_policy_engine_factory(self, factory: Any) -> None:
        """Register a zero-arg factory that builds the PolicyEngine on first
        use. Deferred so constructing the gateway does not spin up the Prolog
        (janus) engine until governance is actually evaluated.

        Args:
            factory: ``() -> PolicyEngine``.
        """
        self._engine_factory = factory

    @traces(spec="014", story="US-003", reqs=["REQ-014-041"])
    def _resolve_engine(self) -> Any:
        """Return the PolicyEngine, building it once from the factory on first
        use. Returns None when no engine/factory is configured (stub mode)."""
        if self._policy_engine is None and self._engine_factory is not None:
            try:
                self._policy_engine = self._engine_factory()
            except Exception as exc:
                logger.warning("policy engine init failed: %s", exc)
                self._engine_factory = None  # don't retry on every call
        return self._policy_engine

    @traces(spec="014", story="US-003", reqs=["REQ-014-031"])
    async def handle_decisions(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle governance.decisions -- subscribe to decisions.

        Args:
            connection_id: Caller connection ID.
            params: Optional filters.

        Returns:
            subscription_id and topic.
        """
        filters = params.get("filters", {})
        sub_id = await self._pubsub.subscribe(
            connection_id,
            "governance.decisions",
            filters,
        )
        return {
            "subscription_id": sub_id,
            "topic": "governance.decisions",
        }

    @traces(spec="014", story="US-003", reqs=["REQ-014-036"])
    async def handle_query(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle governance.query -- historical governance decisions.

        Reads recent PolicyDecision nodes from the graph audit trail
        (written by GraphAuditTrail.log_policy_decision as
        ``(:PolicyDecision {data: <json>, timestamp, session_id})``).

        Args:
            connection_id: Caller connection ID.
            params: optional ``status`` filter and ``limit``.

        Returns:
            List of recent decisions, most recent first.
        """
        if not self._graph_repository:
            return {"decisions": []}

        try:
            limit = min(int(params.get("limit", 50)), 1000)
            status = params.get("status")
            query = (
                "MATCH (d:PolicyDecision) "
                "RETURN d.data AS data, d.timestamp AS timestamp "
                "ORDER BY d.timestamp DESC LIMIT $limit"
            )
            decisions = await asyncio.wait_for(
                self._query_decisions(query, {"limit": limit}, status),
                timeout=10.0,
            )
            return {"decisions": decisions}
        except TimeoutError:
            return {"decisions": [], "error": "Query timeout"}
        except Exception as exc:
            return {"decisions": [], "error": str(exc)}

    @traces(spec="014", story="US-003", reqs=["REQ-014-036"])
    async def _query_decisions(
        self,
        query: str,
        params: dict[str, Any],
        status: str | None,
    ) -> list[dict[str, Any]]:
        """Run the decision query and deserialize each node's JSON ``data``."""
        decisions: list[dict[str, Any]] = []
        rows = await self._graph_repository.execute_read(query, params)
        for row in rows:
            raw = row["data"]
            try:
                decision = json.loads(raw) if isinstance(raw, str) else dict(raw or {})
            except json.JSONDecodeError, TypeError, ValueError:
                decision = {"data": raw}
            if row["timestamp"] is not None:
                decision.setdefault("timestamp", row["timestamp"])
            if status and status not in (
                decision.get("status"),
                decision.get("result"),
                decision.get("decision"),
            ):
                continue
            decisions.append(decision)
        return decisions

    @traces(spec="014", story="US-003", reqs=["REQ-014-041"])
    async def handle_evaluate(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle governance.evaluate -- dry-run check via the PolicyEngine.

        When a PolicyEngine is wired, evaluates the intent for real (fail-closed:
        any evaluation error denies). Without an engine, returns the historical
        permissive stub so the gateway still answers in degraded mode.

        Args:
            connection_id: Caller connection ID.
            params: ``intent`` to evaluate; optional ``actor`` and ``context``.

        Returns:
            Evaluation result with ``result`` (allowed/denied) and ``reason``.
        """
        intent = params.get("intent", "")
        engine = self._resolve_engine()
        if engine is None:
            return {
                "intent": intent,
                "result": "allowed",
                "reason": "stub: no governance engine connected",
            }

        raw_context = params.get("context")
        context = dict(raw_context) if isinstance(raw_context, dict) else {}
        actor = context.setdefault("actor", params.get("actor", connection_id))
        try:
            decision = await engine.evaluate(intent, context)
        except Exception as exc:  # fail-closed: deny on evaluation error
            logger.warning("governance evaluate failed: %s", exc)
            await self.publish_decision_event(
                self._decision_event_payload(
                    actor=actor,
                    action=intent,
                    allowed=False,
                    reason=f"evaluation error: {exc}",
                )
            )
            return {
                "intent": intent,
                "result": "denied",
                "reason": f"evaluation error: {exc}",
            }
        await self.publish_decision_event(
            self._decision_event_payload(
                actor=decision.actor,
                action=intent,
                allowed=decision.allowed,
                reason=decision.reason,
                articles_checked=list(decision.articles_checked),
                duration_ms=decision.duration_ms,
                metadata=dict(decision.metadata) if decision.metadata else None,
                timestamp=(
                    decision.timestamp.isoformat() if decision.timestamp else None
                ),
            )
        )
        return {
            "intent": intent,
            "result": "allowed" if decision.allowed else "denied",
            "reason": decision.reason,
            "decision_type": getattr(
                decision.decision_type, "name", str(decision.decision_type)
            ),
            "articles_checked": list(decision.articles_checked),
        }

    @staticmethod
    @traces(spec="014", story="US-003", reqs=["REQ-014-041"])
    def _decision_event_payload(
        *,
        actor: str,
        action: str,
        allowed: bool,
        reason: str,
        articles_checked: list[int] | None = None,
        duration_ms: float = 0.0,
        metadata: dict[str, Any] | None = None,
        timestamp: str | None = None,
    ) -> dict[str, Any]:
        """Build the ``governance.events`` decision payload shape consumed by
        the VizInt audit-trail and notification-center gadgets
        (``{decision: {decision_id, timestamp, actor, action, resource,
        result, latency_ms, article_violation, details}}``).

        Args:
            actor: The subject the decision was evaluated for.
            action: The intent/action string that was evaluated.
            allowed: Whether the decision permits the action.
            reason: Human-readable rationale from the engine.
            articles_checked: Constitutional article numbers consulted.
            duration_ms: Evaluation latency in milliseconds.
            metadata: Additional engine-supplied decision metadata.
            timestamp: ISO-8601 timestamp; defaults to now.

        Returns:
            The decision dict (unwrapped -- callers nest it under
            ``{"decision": ...}`` via :meth:`publish_decision_event`).
        """
        articles = list(articles_checked or [])
        return {
            "decision_id": str(uuid.uuid4()),
            "timestamp": timestamp or datetime.now(UTC).isoformat(),
            "actor": actor,
            "action": action,
            "resource": "",
            "result": "allowed" if allowed else "denied",
            "latency_ms": duration_ms,
            "article_violation": articles[0] if (not allowed and articles) else None,
            "details": dict(metadata or {}),
        }

    @traces(spec="014", story="US-003", reqs=["REQ-014-031"])
    async def publish_decision_event(self, decision_payload: dict[str, Any]) -> None:
        """Broadcast a real governance decision to ``governance.events``
        subscribers.

        Delivers through the message processor's real broadcast (the same
        path ``volk.lifecycle``/``session.events``/``cxp.traffic`` use), so
        the audit-trail and notification-center gadgets' real-time
        subscriptions actually receive it. A no-op when no processor has
        been injected yet (e.g. constructed before gateway boot wires it via
        ``set_processor``) -- mirrors ``VolkHandler.publish_lifecycle_event``.

        Args:
            decision_payload: The decision dict built by
                :meth:`_decision_event_payload`.
        """
        if self._processor is None:
            return
        await self._processor.broadcast(
            "governance.events", {"decision": decision_payload}
        )

    @traces(spec="014", story="US-003", reqs=["REQ-014-046"])
    async def handle_rules(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle governance.rules -- the active constitutional rule set.

        When a PolicyEngine is wired, returns the constitutional articles the
        engine enforces; otherwise an empty list (degraded mode).

        Args:
            connection_id: Caller connection ID.
            params: Ignored.

        Returns:
            List of rules (constitutional articles).
        """
        if self._policy_engine is None and self._engine_factory is None:
            return {"rules": []}

        # The constitutional articles are static — no need to build (spin up
        # janus) the engine just to list the active rule set.
        from orchestrata_ultimate.governance.constitutional import ARTICLES

        rules = [
            {
                "number": article.number,
                "title": article.title,
                "description": article.description,
                "priority": getattr(article.priority, "name", str(article.priority)),
                "predicates": list(article.predicates),
            }
            for article in ARTICLES
        ]
        return {"rules": rules}