orchestrata_ultimate.governance.policy_engine

Reference › orchestrata_ultimate › governance › policy_engine

The central decision point of the governance package. PolicyEngine implements IPolicyEngine from interfaces and answers one question: may this actor perform this action in this context? It returns a typed PolicyDecision from types. It composes nearly every other module of the package and is serves SPEC-007 (REQ-007-FR10-033, REQ-007-US-001-003, -008, REQ-007-US-002-002). event_emitting_policy_engine wraps it to broadcast decisions. The curated governance page gives the narrative.

The constructor takes a GovernancePrologClient and optional collaborators: a cache, a constitutional enforcer, an ethical boundary, a layered policy client, monitoring, audit, safe mode and a violation tracker. It creates the constitutional, ethical and layered ones itself when not given.

evaluate(action, context) runs a fixed pipeline and returns at the first step that decides:

1. _check_safe_mode denies with safe_mode_active when safe mode is on and the action is not on its allowed list.
2. _check_cache returns a cached decision for a key made from the action and context.
3. _check_constitutional runs the critical constitutional articles and denies with the failed article numbers. If that check raises, strict mode (see _strict) turns the error into a denial.
4. _check_ethical denies on an ethical violation.
5. _check_policy_layers consults the layered rules (SPEC-033 B27) and can only narrow: a layered deny becomes a denial naming the deciding layer, while allow, silent or an error mean no opinion.
6. _evaluate_prolog_and_finalise asks may_execute of the Prolog client. A Prolog error becomes a denial with the reason evaluation_error. An allowed decision is cached; a denial is passed to _handle_violation.

_handle_violation records a Violation (high severity if the reason mentions a constitutional failure), counts it in monitoring and asks safe mode to check the violation rate. evaluate_batch evaluates a list of actions, is_allowed returns just the boolean, and _record writes metrics and the audit entry.

At a glance

Class diagram

Dependency diagram

API

Classes

PolicyEngine

Bases: IPolicyEngine · Decorators: traces(spec='007', story='US-PGOV-001', reqs=['REQ-007-US-001-001', 'REQ-007-FR1-003', 'REQ-007-FR1-006', 'REQ-007-US-0… · Line 47

Central governance policy engine.

Source

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

# mypy: ignore-errors
"""Central governance policy engine for Orchestrata.

Wires together all governance components into a unified IPolicyEngine.
"""

import logging
import time
from collections.abc import Mapping
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance._strict import governance_strict
from orchestrata_ultimate.governance.audit import GovernanceAudit
from orchestrata_ultimate.governance.cache import GovernanceCache, generate_cache_key
from orchestrata_ultimate.governance.constitutional import ConstitutionalEnforcer
from orchestrata_ultimate.governance.ethical import EthicalBoundary
from orchestrata_ultimate.governance.interfaces import IPolicyEngine
from orchestrata_ultimate.governance.monitoring import GovernanceMonitoring
from orchestrata_ultimate.governance.policy_layers import LayeredPolicyClient
from orchestrata_ultimate.governance.prolog_client import GovernancePrologClient
from orchestrata_ultimate.governance.safe_mode import SafeModeManager
from orchestrata_ultimate.governance.types import (
    DecisionType,
    PolicyDecision,
    Violation,
    ViolationType,
)
from orchestrata_ultimate.governance.violations import ViolationTracker

_logger = logging.getLogger(__name__)


@traces(
    spec="007",
    story="US-PGOV-001",
    reqs=[
        "REQ-007-US-001-001",
        "REQ-007-FR1-003",
        "REQ-007-FR1-006",
        "REQ-007-US-001-003",
        "REQ-007-US-001-008",
        "REQ-007-US-001-009",
        "REQ-007-US-001-011",
    ],
)
class PolicyEngine(IPolicyEngine):
    """Central governance policy engine.

    Orchestrates constitutional, ethical, resource, and Prolog-based
    policy evaluation with caching, monitoring, and audit.
    """

    def __init__(
        self,
        prolog_client: GovernancePrologClient,
        cache: GovernanceCache | None = None,
        constitutional: ConstitutionalEnforcer | None = None,
        ethical: EthicalBoundary | None = None,
        policy_layers: LayeredPolicyClient | None = None,
        monitoring: GovernanceMonitoring | None = None,
        audit: GovernanceAudit | None = None,
        safe_mode: SafeModeManager | None = None,
        violations: ViolationTracker | None = None,
        env: Mapping[str, str] | None = None,
    ) -> None:
        self._prolog_client = prolog_client
        self._cache = cache
        self._constitutional = constitutional or ConstitutionalEnforcer(prolog_client)
        self._ethical = ethical or EthicalBoundary(prolog_client)
        # SPEC-033 B27 (ADR-0009): the defeasible policy-layering scheme.
        # Reuses prolog_client's own bridge/engine (no second Prolog engine).
        self._policy_layers = policy_layers or LayeredPolicyClient(prolog_client.bridge)
        self._policy_layers_consulted = False
        self._monitoring = monitoring
        self._audit = audit
        self._safe_mode = safe_mode
        self._violations = violations
        self._env = env

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-033",
            "REQ-007-US-001-003",
            "REQ-007-US-001-008",
            "REQ-007-US-002-002",
        ],
    )
    async def evaluate(self, action: str, context: dict[str, Any]) -> PolicyDecision:
        """Evaluate an action against all governance policies.

        Pipeline:
        1. Check safe mode
        2. Check cache
        3. Constitutional check (critical articles)
        4. Ethical check
        4b. Defeasible policy-layer check (SPEC-033 B27, narrowing-only)
        5. Prolog policy evaluation
        6. Cache result
        7. Record metrics and audit
        """
        start = time.monotonic()
        actor = context.get("actor", context.get("volk_id", "unknown"))

        # 1. Safe mode check
        safe_decision = await self._check_safe_mode(action, actor, start)
        if safe_decision is not None:
            return safe_decision

        # 2. Cache check
        cache_key = generate_cache_key(action, context)
        cached_decision = await self._check_cache(cache_key, start)
        if cached_decision is not None:
            return cached_decision

        # 3. Constitutional check (critical articles first)
        articles_checked: list[int] = []
        const_decision, articles_checked = await self._check_constitutional(
            action, actor, context, articles_checked, start
        )
        if const_decision is not None:
            return const_decision

        # 4. Ethical check
        ethical_decision = await self._check_ethical(
            action, actor, context, articles_checked, start
        )
        if ethical_decision is not None:
            return ethical_decision

        # 4b. Defeasible policy-layer check (SPEC-033 B27, narrowing-only)
        layers_decision = await self._check_policy_layers(
            action, actor, context, articles_checked, start
        )
        if layers_decision is not None:
            return layers_decision

        # 5-7. Prolog evaluation, cache, record
        return await self._evaluate_prolog_and_finalise(
            action, actor, context, articles_checked, cache_key, start
        )

    @traces(spec="007", story="FR-PGOV-008", reqs=["REQ-007-FR8-001"])
    async def _check_safe_mode(
        self, action: str, actor: str, start: float
    ) -> PolicyDecision | None:
        if self._safe_mode is not None and self._safe_mode.is_safe_mode():
            if not self._safe_mode.is_action_allowed_in_safe_mode(action):
                decision = PolicyDecision(
                    decision_type=DecisionType.DENY,
                    action=action,
                    actor=actor,
                    allowed=False,
                    reason="safe_mode_active",
                )
                await self._record(decision, time.monotonic() - start)
                return decision
        return None

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-005"])
    async def _check_cache(self, cache_key: str, start: float) -> PolicyDecision | None:
        if self._cache is not None:
            cached = self._cache.get(cache_key)
            if cached is not None:
                cached.cache_hit = True
                cached.duration_ms = (time.monotonic() - start) * 1000
                await self._record(cached, time.monotonic() - start)
                return cached
        return None

    @traces(spec="007", story="US-PGOV-002", reqs=["REQ-007-US-002-008"])
    async def _check_constitutional(
        self,
        action: str,
        actor: str,
        context: dict[str, Any],
        articles_checked: list[int],
        start: float,
    ) -> tuple[PolicyDecision | None, list[int]]:
        try:
            const_result = await self._constitutional.check_critical(action, context)
            articles_checked = (
                const_result.articles_passed + const_result.articles_failed
            )
            if not const_result.compliant:
                decision = PolicyDecision(
                    decision_type=DecisionType.DENY,
                    action=action,
                    actor=actor,
                    allowed=False,
                    reason=f"constitutional_violation:articles={const_result.articles_failed}",
                    articles_checked=articles_checked,
                )
                await self._handle_violation(decision)
                await self._record(decision, time.monotonic() - start)
                return decision, articles_checked
        except Exception as exc:
            _logger.error("Constitutional check error: %s", exc)
            if governance_strict(self._env):
                decision = PolicyDecision(
                    decision_type=DecisionType.DENY,
                    action=action,
                    actor=actor,
                    allowed=False,
                    reason=f"Constitutional check failed: {exc}",
                    articles_checked=articles_checked,
                )
                await self._record(decision, time.monotonic() - start)
                return decision, articles_checked
        return None, articles_checked

    @traces(spec="007", story="FR-PGOV-005", reqs=["REQ-007-FR1-006"])
    async def _check_ethical(
        self,
        action: str,
        actor: str,
        context: dict[str, Any],
        articles_checked: list[int],
        start: float,
    ) -> PolicyDecision | None:
        try:
            ethical_result = await self._ethical.check(action, context)
            if not ethical_result.ethical:
                decision = PolicyDecision(
                    decision_type=DecisionType.DENY,
                    action=action,
                    actor=actor,
                    allowed=False,
                    reason=f"ethical_violation:{','.join(ethical_result.concerns)}",
                    articles_checked=articles_checked,
                )
                await self._handle_violation(decision)
                await self._record(decision, time.monotonic() - start)
                return decision
        except Exception as exc:
            _logger.error("Ethical check error: %s", exc)
        return None

    async def _check_policy_layers(
        self,
        action: str,
        actor: str,
        context: dict[str, Any],
        articles_checked: list[int],
        start: float,
    ) -> PolicyDecision | None:
        """SPEC-033 B27 defeasible policy layering (ADR-0009) — narrowing-only.

        Consults ``policy_layers.pl``'s ``layered_verdict/3`` (constitutional
        > ethical > resource > team_norms — the highest-priority layer with
        an opinion wins). A ``deny`` verdict becomes a DENY here, named after
        its deciding layer. ``allow`` and ``silent`` both return None ("no
        opinion, keep going"): this step may only ever NARROW what the rest
        of the pipeline decides, never widen it — an allow here must not
        short-circuit a later Prolog deny, and per ``LayeredVerdict``'s
        docstring ``silent`` is explicitly NOT allow, so it degrades to "no
        opinion" exactly like an unreachable/unconsulted rule file rather
        than granting access itself.
        """
        try:
            if not self._policy_layers_consulted:
                if not self._policy_layers.consult():
                    _logger.error(
                        "policy_layers.pl could not be consulted — "
                        "policy-layer check has no opinion"
                    )
                    return None
                self._policy_layers_consulted = True
            context_key = str(context.get("session_id", "unknown"))
            verdict = self._policy_layers.layered_verdict(action, context_key)
        except Exception as exc:
            _logger.error("Policy-layer evaluation error: %s", exc)
            return None

        if verdict.verdict != "deny":
            return None

        decision = PolicyDecision(
            decision_type=DecisionType.DENY,
            action=action,
            actor=actor,
            allowed=False,
            reason=f"policy_layer_violation:{verdict.deciding_layer}",
            articles_checked=articles_checked,
        )
        await self._handle_violation(decision)
        await self._record(decision, time.monotonic() - start)
        return decision

    @traces(
        spec="001", story="US-003", reqs=["REQ-001-US-003-008", "REQ-001-US-003-005"]
    )
    async def _evaluate_prolog_and_finalise(
        self,
        action: str,
        actor: str,
        context: dict[str, Any],
        articles_checked: list[int],
        cache_key: str,
        start: float,
    ) -> PolicyDecision:
        # 5. Prolog policy evaluation
        try:
            allowed = self._prolog_client.may_execute(action, actor, context)
            decision_type = DecisionType.ALLOW if allowed else DecisionType.DENY
            reason = "policy_allow" if allowed else "policy_deny"
        except Exception as exc:
            _logger.error("Prolog evaluation error: %s", exc)
            decision_type = DecisionType.DENY
            allowed = False
            reason = f"evaluation_error:{exc}"

        duration_ms = (time.monotonic() - start) * 1000
        decision = PolicyDecision(
            decision_type=decision_type,
            action=action,
            actor=actor,
            allowed=allowed,
            reason=reason,
            articles_checked=articles_checked,
            duration_ms=duration_ms,
        )

        # 6. Cache result
        if self._cache is not None and allowed:
            self._cache.set(cache_key, decision)

        # 7. Record metrics and audit
        await self._record(decision, time.monotonic() - start)

        if not allowed:
            await self._handle_violation(decision)

        return decision

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-018",
            "REQ-007-US-001-004",
        ],
    )
    async def evaluate_batch(
        self, actions: list[dict[str, Any]]
    ) -> list[PolicyDecision]:
        """Evaluate multiple actions in batch."""
        results = []
        for action_spec in actions:
            action = action_spec.get("action", "unknown")
            context = action_spec.get("context", {})
            result = await self.evaluate(action, context)
            results.append(result)
        return results

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-033",
            "REQ-007-US-001-008",
        ],
    )
    async def is_allowed(self, action: str, context: dict[str, Any]) -> bool:
        """Quick check if an action is allowed."""
        decision = await self.evaluate(action, context)
        return decision.allowed

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=["REQ-007-FR1-007", "REQ-007-US-001-006"],
    )
    async def _record(self, decision: PolicyDecision, elapsed: float) -> None:
        """Record metrics and audit for a decision."""
        duration_ms = elapsed * 1000
        decision.duration_ms = duration_ms

        if self._monitoring is not None:
            self._monitoring.record_query(duration_ms, decision.cache_hit)

        if self._audit is not None:
            try:
                await self._audit.log_decision(decision)
            except Exception as exc:
                _logger.debug("Audit log failed: %s", exc)

    @traces(spec="007", story="US-PGOV-002", reqs=["REQ-007-US-002-004"])
    async def _handle_violation(self, decision: PolicyDecision) -> None:
        """Handle a policy violation."""
        if self._violations is not None:
            import uuid

            violation = Violation(
                id=str(uuid.uuid4()),
                type=ViolationType.POLICY,
                action=decision.action,
                actor=decision.actor,
                severity="high" if "constitutional" in decision.reason else "medium",
            )
            await self._violations.record(violation)

        if self._monitoring is not None:
            self._monitoring.record_violation()

        if self._safe_mode is not None and self._violations is not None:
            if self._violations.should_trigger_safe_mode():
                rate = self._violations.get_rate()
                self._safe_mode.check_violation_rate(rate)