orchestrata_ultimate.governance.verdict

Reference › orchestrata_ultimate › governance › verdict

Attribution for governance decisions (W1-P6). When a governance surface answers allow or deny, this module records where the answer came from, so that a real policy decision is never mistaken for a fallback after an error. prolog_client, prolog_gate and the kernel policy_evaluator import it. It imports nothing from the project.

Provenance has two values: POLICY (a real Prolog rule fired, whether it matched or genuinely failed) and ERROR (the query errored or timed out and a default decided). GovernanceVerdict is a frozen record of the surface name, decision, provenance, rule source and reason, with source_label giving POLICY_ALLOW, POLICY_DENY, ERROR_ALLOW or ERROR_DENY. make_verdict builds one and writes a structured log line starting governance_verdict surface=... source=...; error verdicts log at warning level, the wiring-regression signal, and policy verdicts at info level. explain_may_execute_deny and explain_policy_permits_deny take a boolean query function and return a short text saying why a may_execute or policy_permits deny happened, which the callers append to the reason.

At a glance

Class diagram

Dependency diagram

API

Classes

Provenance (enum)

Bases: Enum · Line 52

Where a governance decision came from.

GovernanceVerdict (dataclass)

Decorators: dataclass(frozen=True) · Line 60

One attributable governance decision (W1-P6 deliverable 1).

Functions

Source

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

"""Verdict provenance for governance enforcement surfaces (W1-P6).

Every governance surface that decides ALLOW/DENY must be attributable:
did a real Prolog rule fire (``POLICY``), or did an error/timeout/missing
predicate cause a default to be applied (``ERROR``)? The G-D1 deny-by-default
flip (see ``reports/proposal-wire-prolog-governance.md`` §7 Phase 6) is
gated on "ERROR_ALLOW == 0 with all POLICY_DENYs attributable" over a
validation window -- this module is the instrument that measurement needs.

The four verdict sources this repo cares about:
    POLICY_ALLOW -- a real rule matched and the answer was allow.
    POLICY_DENY  -- a real rule matched (or genuinely had no solutions) and
                    the answer was deny.
    ERROR_ALLOW  -- a query errored/timed out and the ERROR-path default
                    happened to be allow. Forbidden under strict mode; a
                    sudden non-zero count of these is the "silent no-op"
                    alarm (the exact failure class of incidents 1-3 in the
                    proposal).
    ERROR_DENY   -- a query errored/timed out and the ERROR-path default
                    (fail-closed) denied. Expected under strict mode; a
                    sustained non-zero count is the "wiring regression"
                    alarm even though the immediate effect is safe.

Classifying POLICY vs ERROR: ``PrologBridge.query`` (orchestrator/
prolog_bridge.py) already catches every exception internally and returns
``QueryResult(success=False, error=<message>)`` -- so ``error is not None``
is the ONLY reliable signal that the query itself misbehaved (undefined
predicate, janus exception, syntax error). A ``success=False`` result with
``error is None`` is a genuine Prolog failure -- no clause matched -- which
IS a real policy verdict (POLICY_DENY), not an error. Call sites that use a
different Prolog surface (``core.prolog_adapter.PrologEngine``, which CAN
raise) classify the same way: an exception means ERROR, a query that
executes and returns no solutions means POLICY_DENY.

Backward compatibility: every governance method keeps its existing bool/
enum return shape. This module only adds an ADDITIONAL structured log line
plus (where an existing dataclass result is extended) an extra
``provenance``/``rule_source`` field with a safe default, per the brief's
"least invasive seam" instruction -- no existing caller's contract changes.
"""

from __future__ import annotations

import logging
from collections.abc import Callable
from dataclasses import dataclass
from enum import Enum

_verdict_logger = logging.getLogger("orchestrata_ultimate.governance.verdict")


class Provenance(Enum):
    """Where a governance decision came from."""

    POLICY = "policy"  # a real Prolog rule fired (matched, or genuinely failed)
    ERROR = "error"  # the query errored/timed out; a fallback default decided


@dataclass(frozen=True)
class GovernanceVerdict:
    """One attributable governance decision (W1-P6 deliverable 1)."""

    surface: str
    decision: str
    provenance: Provenance
    rule_source: str
    reason: str

    def source_label(self, is_deny: bool) -> str:
        """``POLICY_ALLOW`` / ``POLICY_DENY`` / ``ERROR_ALLOW`` / ``ERROR_DENY``."""
        prefix = "ERROR" if self.provenance is Provenance.ERROR else "POLICY"
        return f"{prefix}_{'DENY' if is_deny else 'ALLOW'}"


def make_verdict(
    *,
    surface: str,
    decision: str,
    is_deny: bool,
    is_error: bool,
    rule_source: str,
    reason: str,
    logger_: logging.Logger | None = None,
) -> GovernanceVerdict:
    """Build a :class:`GovernanceVerdict` and emit its structured log line.

    The log line is what the P6 validation window aggregates
    (``governance_verdict surface=... source=POLICY_DENY|...``). ``ERROR``
    provenance logs at WARNING (it is the wiring-regression signal); a real
    ``POLICY`` verdict logs at INFO.
    """
    verdict = GovernanceVerdict(
        surface=surface,
        decision=decision,
        provenance=Provenance.ERROR if is_error else Provenance.POLICY,
        rule_source=rule_source,
        reason=reason,
    )
    log = logger_ or _verdict_logger
    level = logging.WARNING if is_error else logging.INFO
    log.log(
        level,
        "governance_verdict surface=%s source=%s decision=%s rule_source=%s reason=%s",
        verdict.surface,
        verdict.source_label(is_deny),
        verdict.decision,
        verdict.rule_source,
        verdict.reason,
    )
    return verdict


def _atom(value: str) -> str:
    """Format a Python string as a safe single-quoted Prolog atom."""
    return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"


# ---------------------------------------------------------------------------
# Proof-tree explanations (W1-P6 deliverable 3 / Prolog#2 proposal item 2)
# ---------------------------------------------------------------------------
#
# janus-swi has no built-in proof-tree API. governance.pl's decision
# predicates (``may_execute/2``, ``policy_permits/4``) are if-then-else /
# negation-as-failure chains over a small, STABLE set of dynamic facts
# (volk_operational/1, volk_exists/1, agent_action_blocked/2,
# volk_blacklisted/1, volk_active/1) -- a generic vanilla meta-interpreter
# would need to faithfully reproduce ``\+`` and ``->`` control constructs to
# avoid MISreporting which leg failed (a correctness risk for a governance
# explainability feature, of all things). Because the predicate set is small
# and does not change often, querying each leg directly in Python and
# reporting which one is responsible is more robust than shipping a second,
# hand-rolled Prolog interpreter for this. This is the documented choice
# ("structured decomposition in Python") the brief names as an alternative
# to a ``prolog/explain.pl`` meta-interpreter.


def explain_may_execute_deny(
    query_bool: Callable[[str], bool], agent: str, action: str
) -> str:
    """Decompose a ``governance:may_execute/2`` DENY into its dependent legs
    (governance.pl:843-865): ``volk_operational/1``, ``volk_exists/1``,
    ``agent_action_blocked/2``.

    Args:
        query_bool: A callable that evaluates one Prolog goal string and
            returns whether it has a solution (e.g. ``PrologBridge.query_bool``
            or an equivalent wrapper over ``PrologEngine.query``).
        agent: The agent/actor that was denied.
        action: The action that was denied.
    """
    agent_atom, action_atom = _atom(agent), _atom(action)
    operational = query_bool(f"governance:volk_operational({agent_atom})")
    exists = query_bool(f"governance:volk_exists({agent_atom})")
    blocked = query_bool(
        f"governance:agent_action_blocked({agent_atom}, {action_atom})"
    )
    legs = (
        f"volk_operational({agent})={operational}, "
        f"volk_exists({agent})={exists}, "
        f"agent_action_blocked({agent},{action})={blocked}"
    )
    if blocked:
        return (
            f"proof: agent_action_blocked({agent},{action}) succeeded "
            f"(explicit block) [{legs}]"
        )
    if exists and not operational:
        return (
            f"proof: {agent} is a registered volk (volk_exists) but not "
            f"operational, and clause 2's \\+ volk_exists guard also fails "
            f"-- no may_execute/2 clause matches [{legs}]"
        )
    return f"proof: no may_execute/2 clause matched for ({agent}, {action}) [{legs}]"


def explain_policy_permits_deny(
    query_bool: Callable[[str], bool],
    from_volk: str,
    to_volk: str,
    message_type: str,
) -> str:
    """Decompose a ``policy_permits/4`` (routing) DENY into its dependent
    legs (governance.pl:281-293): ``volk_exists/1`` (both sides),
    ``volk_blacklisted/1``, ``agent_action_blocked/2``."""
    from_atom, to_atom, msg_atom = _atom(from_volk), _atom(to_volk), _atom(message_type)
    exists_from = query_bool(f"governance:volk_exists({from_atom})")
    exists_to = query_bool(f"governance:volk_exists({to_atom})")
    blacklisted_to = query_bool(f"governance:volk_blacklisted({to_atom})")
    blocked = query_bool(
        f"governance:agent_action_blocked({from_atom}, route({to_atom}, {msg_atom}))"
    )
    legs = (
        f"volk_exists({from_volk})={exists_from}, "
        f"volk_exists({to_volk})={exists_to}, "
        f"volk_blacklisted({to_volk})={blacklisted_to}, "
        f"agent_action_blocked({from_volk},route(...))={blocked}"
    )
    if not exists_from:
        return (
            f"proof: sender {from_volk} is not a known volk (\\+ volk_exists) [{legs}]"
        )
    if not exists_to:
        return (
            f"proof: recipient {to_volk} is not a known volk (\\+ volk_exists) "
            f"[{legs}]"
        )
    if blacklisted_to:
        return f"proof: recipient {to_volk} is volk_blacklisted [{legs}]"
    if blocked:
        return (
            f"proof: agent_action_blocked({from_volk}, "
            f"route({to_volk},{message_type})) succeeded [{legs}]"
        )
    return (
        f"proof: policy_permits/4 had no allow/queue clause for "
        f"({from_volk}->{to_volk}, {message_type}) [{legs}]"
    )