orchestrata_ultimate.governance.policy_layers

Reference › orchestrata_ultimate › governance › policy_layers

The client for the defeasible policy layers of SPEC-033 B27 (ADR-0009). Several rule layers may have an opinion about an action; the layers run in priority order (constitutional, ethical, resource, team norms) and the highest layer with an opinion wins. policy_engine is the importer and uses the result only to narrow its decision.

LayeredVerdict is a frozen record with a verdict of allow, deny or silent and the name of the deciding layer. The docstring is explicit that silent (every layer had no opinion) is not allow, and leaves the caller's own default in force. LayeredPolicyClient wraps a PrologBridge. consult loads policy_layers.pl only; its clauses refer to the constitutional, ethics and resource modules at call time, so those must already be loaded, for example with load_governance_rules in prolog_client. layered_verdict(action, context) queries policy_layers:layered_verdict/3, returns silent when there is no solution, otherwise decodes the verdict, finds the deciding layer with _find_deciding_layer and raises ValueError for any verdict other than allow or deny. assert_team_norm adds a team-norm fact for an action, context key and verdict.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

LayeredVerdict (dataclass)

Decorators: dataclass(frozen=True) · Line 51

The result of one layered_verdict/3 evaluation.

LayeredPolicyClient

Line 69

Thin real-Prolog wrapper around policy_layers.pl.

Source

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

"""Python client for the SPEC-033 B27 defeasible policy layering scheme.

Wraps :class:`~orchestrata_ultimate.infra.prolog_bridge.PrologBridge`
with the same conventions as ``governance/prolog_client.py``'s
``GovernancePrologClient`` (module-qualified queries, ``_format_atom`` atom
escaping, a thin real-Prolog wrapper — no business logic lives here, it all
lives in ``prolog/policy_layers.pl``).

Layer order (highest priority first, matches ``policy_layers.pl``'s
``layer_rank/2`` facts): constitutional(0) > ethical(1) > resource(2) >
team_norms(3). ``layered_verdict`` returns the verdict of the
highest-priority layer that is NOT silent (see the module header of
``policy_layers.pl`` for the "lower never widens higher" construction
proof, ADR-0009).
"""

# B27 wiring: consulted and queried by PolicyEngine._check_policy_layers
# (governance/policy_engine.py) as a narrowing-only step between the
# ethical check and the final Prolog evaluation. Not wired into
# PrologGate — that remains a separate, reviewed change.

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Literal

if TYPE_CHECKING:
    from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

from orchestrata_ultimate.core.infra_config import PROLOG_RULES_DIR

_POLICY_LAYERS_PL = "policy_layers.pl"

# Layer names in DECREASING priority — matches policy_layers.pl's
# layer_rank/2 facts exactly (constitutional=0, ethical=1, resource=2,
# team_norms=3). Used ONLY to determine ``deciding_layer`` on the Python
# side; the actual precedence enforcement (verdict VALUE) lives entirely in
# policy_layers.pl's layered_verdict/3, driven generically by layer_rank/2.
_LAYERS_BY_PRIORITY: tuple[str, ...] = (
    "constitutional",
    "ethical",
    "resource",
    "team_norms",
)

_VALID_TEAM_NORM_VERDICTS = ("allow", "deny")


@dataclass(frozen=True)
class LayeredVerdict:
    """The result of one ``layered_verdict/3`` evaluation.

    ``verdict`` is ``"silent"`` when EVERY layer (constitutional, ethical,
    resource, team_norms) was silent for the given (action, context) — i.e.
    the Prolog query ``layered_verdict(Action, Context, Verdict)`` had ZERO
    solutions. **``silent`` is NOT ``allow``**: it means this layering
    scheme has no opinion at all, and the caller's OWN existing default
    posture (fail-open or fail-closed) applies — this client does not, and
    must not be read to, decide that posture. ``deciding_layer`` names
    whichever of ``"constitutional"``/``"ethical"``/``"resource"``/
    ``"team_norms"`` produced the verdict, or ``None`` when silent.
    """

    verdict: Literal["allow", "deny", "silent"]
    deciding_layer: str | None


class LayeredPolicyClient:
    """Thin real-Prolog wrapper around ``policy_layers.pl``.

    ``context`` is treated as an opaque atom/string throughout this client
    (formatted as a quoted Prolog atom) — sufficient for the
    constitutional/ethical/team_norms layers, which either ignore Context
    entirely or use it only as a ground lookup key. The resource layer's
    ``access(ResourceType, Resource, Mode)`` compound Context is NOT
    expressible through this string-only API; callers exercising resource-
    layer decisions query ``policy_layers:layer_verdict/4`` (or
    ``layered_verdict/3``) directly via ``self.bridge.query(...)`` with a
    raw compound term (see the test suite for the pattern) — a deliberate,
    documented scope decision for this deliverable, not an oversight.
    """

    def __init__(self, bridge: PrologBridge) -> None:
        self._bridge = bridge

    @property
    def bridge(self) -> PrologBridge:
        return self._bridge

    def _format_atom(self, value: str) -> str:
        """Format a Python string as a safe, quoted Prolog atom."""
        sanitized = value.replace("\\", "\\\\").replace("'", "\\'")
        return f"'{sanitized}'"

    def consult(self, rules_dir: str | Path | None = None) -> bool:
        """Consult ``policy_layers.pl`` into ``self.bridge``'s engine.

        Does NOT consult constitutional.pl/ethics.pl/resources.pl — this
        file's clauses reference them via module-qualified goals
        (``constitutional:...``, ``ethics:...``, ``resources:...``) that
        resolve at CALL time, so callers must ensure those three are
        already consulted (e.g. via
        ``governance.prolog_client.load_governance_rules``) for
        ``layered_verdict``/``layer_verdict`` queries to succeed rather
        than raise an ``existence_error``.
        """
        directory = Path(rules_dir) if rules_dir is not None else Path(PROLOG_RULES_DIR)
        return self._bridge.load_rules(directory / _POLICY_LAYERS_PL)

    def layered_verdict(self, action: str, context: str) -> LayeredVerdict:
        """Evaluate ``policy_layers:layered_verdict(Action, Context,
        Verdict)`` and report which layer decided.

        Returns ``LayeredVerdict(verdict="silent", deciding_layer=None)``
        when the query has zero solutions (every layer silent).
        """
        action_atom = self._format_atom(action)
        context_atom = self._format_atom(context)
        query_str = (
            f"policy_layers:layered_verdict({action_atom}, {context_atom}, Verdict)"
        )
        result = self._bridge.query(query_str)
        if not result.success or not result.solutions:
            return LayeredVerdict(verdict="silent", deciding_layer=None)

        verdict = self._decode(result.solutions[0].get("Verdict"))
        deciding_layer = self._find_deciding_layer(action_atom, context_atom)
        if verdict not in ("allow", "deny"):
            # policy_layers.pl only ever emits allow/deny — an unexpected
            # value here is a genuine defect upstream, not a case to guess
            # a default for.
            raise ValueError(
                f"layered_verdict/3 returned an unexpected verdict: {verdict!r}"
            )
        return LayeredVerdict(
            verdict=verdict,  # type: ignore[arg-type]
            deciding_layer=deciding_layer,
        )

    def _find_deciding_layer(self, action_atom: str, context_atom: str) -> str | None:
        """Which layer (in priority order) is the first non-silent one for
        this (already-atom-formatted) action/context pair."""
        for layer in _LAYERS_BY_PRIORITY:
            query_str = (
                f"policy_layers:layer_verdict({layer}, {action_atom}, "
                f"{context_atom}, _)"
            )
            if self._bridge.query_bool(query_str):
                return layer
        return None

    @staticmethod
    def _decode(value: object) -> str:
        """janus-swi may return atoms as ``bytes`` — decode to ``str`` like
        the sibling ``GovernancePrologClient`` does for Prolog decisions."""
        if isinstance(value, bytes):
            return value.decode()
        return str(value)

    def assert_team_norm(self, action: str, context_key: str, verdict: str) -> None:
        """Assert a GROUND ``team_norm_verdict(Action, ContextKey, Verdict)``
        fact, MODULE-QUALIFIED into ``policy_layers`` (ADR-0005: facts come
        from outside the rulebase, never from rule side-effects).

        A bare (unqualified) ``assertz`` would land in the ``user`` module
        and never be seen by ``layer_verdict(team_norms, ...)`` — the same
        caveat ``test_may_execute_real_rules.py`` documents for
        ``governance:agent_action_blocked/2``.
        """
        if verdict not in _VALID_TEAM_NORM_VERDICTS:
            raise ValueError(
                f"verdict must be one of {_VALID_TEAM_NORM_VERDICTS}, got {verdict!r}"
            )
        action_atom = self._format_atom(action)
        context_atom = self._format_atom(context_key)
        self._bridge._prolog.assertz(
            f"policy_layers:team_norm_verdict({action_atom}, {context_atom}, {verdict})"
        )