orchestrata_ultimate.governance.prolog_gate

Reference › orchestrata_ultimate › governance › prolog_gate

The small, synchronous Prolog hook that the pipeline calls at runtime. PrologGate answers two questions: may an agent execute an action (may_execute), and may a volk complete a task (may_complete). The kernel pipeline creates one, loads its rules at boot, asks may_execute before accepting an intent, and pipeline_escalation asks may_complete when a human approves. It is separate from the richer async PolicyEngine. It serves SPEC-007 (REQ-007-US-001-001) and SPEC-000 story US-010 (REQ-000-087 to -092).

load_rules creates a PrologEngine and consults the core rule files that exist in the rules directory: governance.pl, constitutional.pl, agents.pl, ethics.pl, pam_rules.pl and completion.pl. It returns False and logs when loading fails. may_execute(agent, action) escapes both names as Prolog atoms and queries may_execute/2, returning a flag and a reason: approved on a solution; blocked when there is no rule (with an explanation from governance.verdict); blocked with the error text when the query fails. When rules were never loaded it returns allowed with the reason "Prolog not loaded — permissive mode". Each verdict is also logged through make_verdict.

may_complete(volk, task_id) queries three completion gates, all_acceptance_criteria_met, all_outputs_verified and no_open_questions, and returns the flag with a tick or cross line per gate; a failing gate query counts as a failure. If the whole check errors, strict mode (default) denies and non-strict mode allows. The gate also holds the facts the pipeline asserts: assert_task_output, assert_open_question, retract_open_question, assert_fact, and the volk state helpers assert_volk_operational, retract_volk_operational and set_volk_active.

At a glance

Class diagram

Dependency diagram

API

Classes

PrologGate

Decorators: traces(spec='000', story='US-007', reqs=['REQ-000-066']) · Line 41

Wraps the Prolog engine (janus-swi via core.prolog_adapter) for governance checks.

Functions

Source

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

# mypy: ignore-errors
"""Prolog governance gate — may_execute and may_complete predicates."""

import logging
import os
from collections.abc import Mapping

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.infra_config import PROLOG_RULES_DIR
from orchestrata_ultimate.core.prolog_adapter import PrologEngine
from orchestrata_ultimate.governance._strict import governance_strict
from orchestrata_ultimate.governance.verdict import (
    explain_may_execute_deny,
    make_verdict,
)
from orchestrata_ultimate.governance.volk_facts import (
    active_goals,
    assert_goals,
    retract_goals,
)

_logger = logging.getLogger(__name__)


def _escape_prolog_atom(value: str) -> str:
    """Escape a string for safe use inside a single-quoted Prolog atom.

    Order matters: backslash MUST be escaped first, otherwise the
    backslashes introduced by the later replacements would themselves
    get re-escaped.
    """
    return (
        value.replace("\\", "\\\\")
        .replace("'", "\\'")
        .replace('"', '\\"')
        .replace("\n", "\\n")
    )


@traces(spec="000", story="US-007", reqs=["REQ-000-066"])
class PrologGate:
    """Wraps the Prolog engine (janus-swi via ``core.prolog_adapter``) for
    governance checks."""

    def __init__(self):
        self.prolog = None
        self.loaded = False
        self.session_facts: list[str] = []

    def load_rules(self) -> bool:
        """Load Prolog governance rules."""
        try:
            self.prolog = PrologEngine()

            # Core governance rules + the may_complete/2 completion rules.
            # completion.pl is CONSULTED (not runtime-asserted) so its
            # variable-bearing clauses load on the janus-swi backend too --
            # janus's query-based assertz cannot assert clauses with Prolog
            # variables, but consult parses them natively. Runtime facts
            # (task_has_output/2, has_open_question/2) are ground and asserted
            # at runtime via assert_fact, which works on both backends.
            rules_dir = PROLOG_RULES_DIR
            core_files = [
                "governance.pl",
                "constitutional.pl",
                "agents.pl",
                "ethics.pl",
                "pam_rules.pl",
                "completion.pl",
            ]
            loaded = 0
            for f in core_files:
                path = os.path.join(rules_dir, f)
                if os.path.exists(path):
                    self.prolog.consult(path)
                    loaded += 1

            self.loaded = True
            _logger.info("Loaded %d Prolog rule files", loaded)
            return True
        except Exception as e:
            _logger.error("Failed to load Prolog rules: %s", e)
            self.loaded = False
            return False

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-001"])
    def may_execute(self, agent: str, action: str) -> tuple[bool, str]:
        """Check if an agent may execute an action.

        W1-P6: this method's return VALUE is unchanged byte-for-byte (some
        callers assert exact reason-string equality) -- provenance is
        recorded as an ADDITIONAL side-effect log line via
        ``governance.verdict.make_verdict``, never by altering what is
        returned.
        """
        if not self.loaded:
            # No rules consulted at all: an ALLOW that carries no rule
            # authority whatsoever -- the textbook ERROR_ALLOW case W1-P6
            # exists to make visible (governance_boot's "0 files loaded"
            # signal is this same failure mode at boot time).
            make_verdict(
                surface="PrologGate.may_execute",
                decision="allow",
                is_deny=False,
                is_error=True,
                rule_source="error:prolog_not_loaded",
                reason="Prolog not loaded — permissive mode",
                logger_=_logger,
            )
            return True, "Prolog not loaded — permissive mode"
        try:
            # self.loaded is only set True after self.prolog is constructed
            # in load_rules() — this documents that invariant for pyright.
            assert self.prolog is not None
            agent_escaped = _escape_prolog_atom(agent)
            action_escaped = _escape_prolog_atom(action)
            query = f"may_execute('{agent_escaped}', '{action_escaped}')"
            results = list(self.prolog.query(query))
            if results:
                make_verdict(
                    surface="PrologGate.may_execute",
                    decision="allow",
                    is_deny=False,
                    is_error=False,
                    rule_source="may_execute/2",
                    reason="approved",
                    logger_=_logger,
                )
                return True, "approved"
            reason = f"blocked — no may_execute rule for ({agent}, {action})"
            explanation = explain_may_execute_deny(self._query_bool, agent, action)
            make_verdict(
                surface="PrologGate.may_execute",
                decision="deny",
                is_deny=True,
                is_error=False,
                rule_source="may_execute/2",
                reason=f"{reason} — {explanation}",
                logger_=_logger,
            )
            return False, reason
        except Exception as e:
            # If the predicate does not exist we treat it as "no rule" and
            # block by default to preserve the loaded-but-empty behavior
            # expected by governance tests.
            _logger.error("may_execute Prolog query failed: %s", e)
            message = str(e).lower()
            if "existence_error" in message or "unknown procedure" in message:
                reason = f"blocked — no may_execute rule for ({agent}, {action})"
            else:
                reason = f"blocked — Prolog query error: {e}"
            make_verdict(
                surface="PrologGate.may_execute",
                decision="deny",
                is_deny=True,
                is_error=True,
                rule_source="error:may_execute/2",
                reason=reason,
                logger_=_logger,
            )
            return False, reason

    def _query_bool(self, goal: str) -> bool:
        """Adapt ``PrologEngine.query`` (returns a list, can raise) to the
        ``Callable[[str], bool]`` shape ``explain_may_execute_deny`` expects
        -- mirrors ``PrologBridge.query_bool``'s swallow-on-error contract
        so a leg query failing never breaks the explanation itself."""
        if self.prolog is None:
            return False
        try:
            return bool(list(self.prolog.query(goal)))
        except Exception:
            return False

    @traces(
        spec="000",
        story="US-010",
        reqs=[
            "REQ-000-087",
            "REQ-000-088",
            "REQ-000-089",
            "REQ-000-090",
            "REQ-000-092",
        ],
    )
    def may_complete(
        self, volk: str, task_id: str, env: Mapping[str, str] | None = None
    ) -> tuple[bool, list[str]]:
        """Check if a volk may complete a task. Returns (allowed, [gate_results]).

        On a Prolog error, fails CLOSED (denies) when governance strict mode
        is on (the default — see ``governance._strict.governance_strict``);
        set ``ORCHESTRATA_GOVERNANCE_STRICT=false`` (or pass ``env=``) to
        restore the permissive legacy behavior.
        """
        gates = []
        if not self.loaded:
            gates.append("✓ Prolog not loaded — permissive mode")
            return True, gates

        try:
            # self.loaded is only set True after self.prolog is constructed
            # in load_rules() — this documents that invariant for pyright.
            assert self.prolog is not None
            volk_escaped = _escape_prolog_atom(volk)
            task_id_escaped = _escape_prolog_atom(task_id)
            # Check each gate individually for reporting
            gate_checks = [
                (
                    "all_acceptance_criteria_met",
                    f"all_acceptance_criteria_met('{volk_escaped}', '{task_id_escaped}')",
                ),
                (
                    "all_outputs_verified",
                    f"all_outputs_verified('{volk_escaped}', '{task_id_escaped}')",
                ),
                (
                    "no_open_questions",
                    f"no_open_questions('{volk_escaped}', '{task_id_escaped}')",
                ),
            ]

            all_passed = True
            for gate_name, query in gate_checks:
                try:
                    results = list(self.prolog.query(query))
                    if results:
                        gates.append(f"✓ {gate_name}")
                    else:
                        gates.append(f"✗ {gate_name}")
                        all_passed = False
                except Exception as e:
                    gates.append(f"✗ {gate_name} (error: {e} — default FAIL)")
                    all_passed = False

            return all_passed, gates
        except Exception as e:
            _logger.error("may_complete Prolog query failed: %s", e)
            if governance_strict(env):
                gates.append(f"✗ Prolog error: {e} — fail-closed")
                return False, gates
            gates.append(f"✓ Prolog error: {e} — permissive fallback")
            return True, gates

    @traces(spec="006", story="US-004", reqs=["REQ-006-US-004-002"])
    # ------------------------------------------------------------------
    # Volk fact sync -- the orchestrator half
    # ------------------------------------------------------------------
    #
    # This gate and GovernancePrologClient run the SAME goals against two
    # different engines (a PrologEngine here, a PrologBridge there), which is
    # why the goals are built in governance.volk_facts rather than written out
    # twice. Without these, `may_execute(alias, capability)` at
    # orchestrator/pipeline_runner.py denies every stage: the volk is a
    # stranger to the rule base, and a stranger is denied.

    def _run_volk_goals(self, goals: tuple[str, ...], what: str, alias: str) -> bool:
        """Run *goals*, best-effort. False if the rule base could not take them."""
        if not self.loaded or self.prolog is None:
            return False
        try:
            for goal in goals:
                list(self.prolog.query(goal))
        except Exception as exc:
            _logger.warning("governance %s failed for %s: %s", what, alias, exc)
            return False
        return True

    def assert_volk_operational(self, alias: str) -> bool:
        """Make *alias* satisfy ``governance:volk_operational/1``."""
        return self._run_volk_goals(assert_goals(alias), "volk-fact assert", alias)

    def retract_volk_operational(self, alias: str) -> bool:
        """Remove every ``volk_*`` fact for *alias*."""
        return self._run_volk_goals(retract_goals(alias), "volk-fact retract", alias)

    def set_volk_active(self, alias: str, active: bool) -> bool:
        """Track a live status transition on ``governance:volk_active/1``."""
        return self._run_volk_goals(
            active_goals(alias, active), "volk_active update", alias
        )

    @traces(spec="000", story="US-007", reqs=["REQ-000-066"])
    def assert_fact(self, fact: str):
        """Assert a Prolog fact into the session."""
        if self.loaded:
            try:
                # self.loaded is only set True after self.prolog is
                # constructed in load_rules() — documents that invariant.
                assert self.prolog is not None
                self.prolog.assertz(fact)
                self.session_facts.append(fact)
            except Exception as e:
                _logger.error("Failed to assert fact: %s — %s", fact, e)

    @traces(spec="000", story="US-010", reqs=["REQ-000-089"])
    def assert_task_output(self, volk: str, task_id: str):
        """Mark that a volk has produced output for a task."""
        volk_escaped = _escape_prolog_atom(volk)
        task_id_escaped = _escape_prolog_atom(task_id)
        self.assert_fact(f"task_has_output('{volk_escaped}', '{task_id_escaped}')")

    @traces(spec="000", story="US-010", reqs=["REQ-000-090"])
    def assert_open_question(self, volk: str, task_id: str):
        """Mark that a volk has open questions for a task."""
        volk_escaped = _escape_prolog_atom(volk)
        task_id_escaped = _escape_prolog_atom(task_id)
        self.assert_fact(f"has_open_question('{volk_escaped}', '{task_id_escaped}')")

    @traces(spec="000", story="US-010", reqs=["REQ-000-090"])
    def retract_open_question(self, volk: str, task_id: str):
        """Remove open question marker."""
        if self.loaded:
            try:
                # self.loaded is only set True after self.prolog is
                # constructed in load_rules() — documents that invariant.
                assert self.prolog is not None
                volk_escaped = _escape_prolog_atom(volk)
                task_id_escaped = _escape_prolog_atom(task_id)
                self.prolog.retract(
                    f"has_open_question('{volk_escaped}', '{task_id_escaped}')"
                )
            except Exception as exc:
                _logger.debug("retract open_question failed: %s", exc)