orchestrata_ultimate.postman.governance

Reference › orchestrata_ultimate › postman › governance

The authority checks for broker operations, SPEC-001 story US-003 (REQ-001-US-003-011, -013, -014, -023) and SPEC-002 stories US-007 and FR-010. PostmanGovernanceManager answers who may send to whom, who may publish to or manage a topic or queue, and who may perform admin operations, and returns an AuthResult (authorized flag, reason, optional escalation path and audit id). Nothing in the package imports it yet. It is the postman's own check, separate from the wider governance package and from the simpler rule inside router.

The decision rules are in Python. check_send_authority allows the orchestrator U:Ox; denies an unknown sender with an escalation path to U:Ox; allows a recipient registered for that sender (register_recipient and unregister_recipient); and otherwise requires a permission named send.<message type> for the sender. check_topic_authority allows U:Ox, lets U:PS create and delete topics, handles publishing in _check_publish_authority, and allows every agent to subscribe. check_queue_authority does the same for queues and check_admin_authority allows U:Ox everything and U:PS the queue and topic operations. Permissions, topic owners and queue owners are read from the Prolog rule base as the properties AGENT_PERMISSIONS, TOPIC_OWNERS and QUEUE_OWNERS. A failed check gets the escalation path ["U:Ox"].

Every decision goes through _make_decision, which also asks the Prolog rule base the corresponding goal in audit-only mode and records whether Prolog agrees; a disagreement is logged as a warning and does not change the result. _log_decision writes the decision to an audit log that get_audit_log can read. The Prolog engine is created and loaded once, under a lock, by _ensure_prolog_ready, and _evaluate_prolog_goal fails closed: any error counts as a denial.

At a glance

Class diagram

Dependency diagram

API

Classes

AuthResult (dataclass)

Decorators: dataclass, traces(spec='002', story='US-007', reqs=['REQ-002-US-007-001']) · Line 51

Result of an authority check operation.

PostmanGovernanceManager

Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-011', 'REQ-001-US-003-013', 'REQ-001-US-003-014', 'REQ-001-US-… · Line 77

Manages governance and authorization for Postman message routing.

Functions

Source

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

"""
Governance Integration Layer for Postman Message Broker.

Integrates Postman with PAM++ governance system to validate message routing
against authorization rules, log governance decisions to graph audit trail,
and provide escalation paths when authority checks fail.

Author: Mohamed
"""

import logging
import os
import threading
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any

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.postman.constants import GOVERNANCE_CACHE_TTL_SECONDS
from orchestrata_ultimate.postman.types import MessageType

# Spec: 002 - Postman Message Broker

# Rule base for _query_prolog — mirrors the static AGENT_PERMISSIONS /
# TOPIC_OWNERS / QUEUE_OWNERS tables below as real Prolog facts + rules.
_POSTMAN_RULES_FILENAME = "postman_governance.pl"

logger = logging.getLogger(__name__)


def _to_prolog_term(value: Any) -> str:
    """Render a Python value as a GROUND Prolog term.

    Strings become single-quoted atoms (escaped); bools become the
    ``true``/``false`` atoms; numbers are rendered as-is. Anything else
    falls back to a quoted atom of its ``str()``.
    """
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return str(value)
    escaped = str(value).replace("\\", "\\\\").replace("'", "\\'")
    return f"'{escaped}'"


@dataclass
@traces(spec="002", story="US-007", reqs=["REQ-002-US-007-001"])
class AuthResult:
    """Result of an authority check operation.

    Attributes:
        authorized: Whether the operation is authorized
        reason: Human-readable explanation of the decision
        escalation_path: Ordered list of agents who can approve escalation
        audit_id: Unique identifier for audit log entry
    """

    authorized: bool
    reason: str
    escalation_path: list[str] | None = None
    audit_id: str | None = None


@traces(
    spec="001",
    story="US-003",
    reqs=[
        "REQ-001-US-003-011",
        "REQ-001-US-003-013",
        "REQ-001-US-003-014",
        "REQ-001-US-003-023",
    ],
)
class PostmanGovernanceManager:
    """Manages governance and authorization for Postman message routing.

    Validates all message routing operations against PAM++ governance rules,
    maintains audit trail of governance decisions, and provides escalation
    paths when operations are not directly authorized.
    """

    # ── permission tables ───────────────────────────────────────────────
    #
    # These were literal Python dicts until 2026-08-21, transcribed from the
    # `agent_permission/2`, `topic_owner/2` and `queue_owner/2` facts in
    # postman_governance.pl. The module comment above called that "mirroring";
    # measured, it was 23 permission rows duplicated VERBATIM across two
    # languages, plus 12 topic owners and 8 queue owners. One copy was
    # consulted into the engine on every governance query and never read; the
    # Python copy decided every answer. Editing either one left the other
    # silently disagreeing, with nothing in the tree able to notice.
    #
    # Constitutional Article II is "Prolog Source of Truth — no hardcoded
    # rules", so the rulebase is now the only place these live. They stay
    # class-level NAMES (read as `manager.AGENT_PERMISSIONS`) because that is
    # the surface six internal call sites and ten existing assertions already
    # use — what changed is where the rows come from, not how they are read.
    #
    # Loaded lazily and cached: `__init__`'s contract is that constructing a
    # manager and never issuing a governance query costs no Prolog startup,
    # and a table read is a query. `_ensure_prolog_ready` raising propagates,
    # matching `_evaluate_prolog_goal`'s fail-CLOSED posture — a governance
    # table that cannot be loaded must not silently answer "no permissions".

    # Each property issues its goal as a LITERAL at the call site, rather than
    # through a shared `_facts(goal)` helper. The helper existed for two
    # commits and check_prolog_boundary.py rejected it: a query whose text is
    # a parameter is a non-literal site, invisible to the gate's AST walk, and
    # a governance-shaped file may not have one without a WHITELIST
    # justification. Whitelisting would have re-created exactly the blindness
    # this change set out to remove — `_prolog_audit_goal` builds predicate
    # names at runtime, which is why this subsystem's Prolog surface was
    # ungated in the first place. Three literals cost two lines and keep every
    # query site checkable.
    #
    # `PrologEngine.query` returns a solution LIST, so an empty table and a
    # failed goal are both `[]`. That is correct here: a rule base with no
    # `agent_permission/2` facts genuinely grants nobody anything, and every
    # caller treats a missing row as "no authority". Unlike
    # `_evaluate_prolog_goal` there is no boolean to invert, so no separate
    # fail-closed branch is needed.
    #
    # Declared HERE, at class level, and not only in `__init__`. These are
    # BARE annotations: PEP 526 records them in `__annotations__` and creates
    # no class attribute, so each instance still gets its own from `__init__`
    # and runtime behaviour is byte-for-byte unchanged. What they remove is a
    # type-checker ordering artefact. The properties below appear BEFORE
    # `__init__` in the class body, so mypy took `self._agent_permissions =
    # table` -- the assignment inside the property -- as the defining one,
    # inferred a non-Optional `dict[str, set[str]]`, and then rejected
    # `__init__`'s own annotated `= None` as BOTH a redefinition and an
    # incompatible assignment. That accounted for 9 of the 14 errors this
    # file reported: `has-type` at each property's `is None` test (the type
    # was still being resolved), plus `no-redef` and `assignment` at each of
    # the three `__init__` lines. One authoritative declaration ahead of
    # every use settles it without moving `__init__` or reordering the
    # properties -- both of which would have changed real code to satisfy a
    # checker, rather than telling the checker what the code already means.
    _agent_permissions: dict[str, set[str]] | None
    _topic_owners: dict[str, str] | None
    _queue_owners: dict[str, str] | None

    @property
    def AGENT_PERMISSIONS(self) -> dict[str, set[str]]:  # noqa: N802
        """``agent_permission(Agent, Permission)`` as ``{agent: {perm, ...}}``."""
        if self._agent_permissions is None:
            engine = self._ensure_prolog_ready()
            table: dict[str, set[str]] = {}
            for row in engine.query("agent_permission(Agent, Permission)"):
                table.setdefault(str(row["Agent"]), set()).add(str(row["Permission"]))
            self._agent_permissions = table
        return self._agent_permissions

    @property
    def TOPIC_OWNERS(self) -> dict[str, str]:  # noqa: N802
        """``topic_owner(Topic, Owner)`` as ``{topic: owner}``."""
        if self._topic_owners is None:
            engine = self._ensure_prolog_ready()
            self._topic_owners = {
                str(row["Topic"]): str(row["Owner"])
                for row in engine.query("topic_owner(Topic, Owner)")
            }
        return self._topic_owners

    @property
    def QUEUE_OWNERS(self) -> dict[str, str]:  # noqa: N802
        """``queue_owner(Queue, Owner)`` as ``{queue: owner}``."""
        if self._queue_owners is None:
            engine = self._ensure_prolog_ready()
            self._queue_owners = {
                str(row["Queue"]): str(row["Owner"])
                for row in engine.query("queue_owner(Queue, Owner)")
            }
        return self._queue_owners

    def __init__(
        self, prolog_engine: Any = None, prolog_rules_path: str | None = None
    ) -> None:
        """Initialize the governance manager.

        Args:
            prolog_engine: Optional pre-configured Prolog engine (janus-swi,
                ``core.prolog_adapter.PrologEngine``) reused for
                ``_query_prolog``. If None (the default), an engine is
                created and the rule base consulted LAZILY on the first
                Prolog query, so the many existing callers that construct
                this manager and never exercise ``_query_prolog`` incur no
                Prolog engine startup cost.
            prolog_rules_path: Optional override for the rule base file path
                (mirrors ``PrologBridge``'s ``cxp_rules_path`` seam). Defaults
                to ``<PROLOG_RULES_DIR>/postman_governance.pl``.
        """
        # Simulated graph storage - will be replaced with real graph client
        self._audit_log: dict[str, dict[str, Any]] = {}

        # Rule cache with TTL
        self._rule_cache: dict[str, tuple[bool, float]] = {}
        self._cache_ttl: float = float(GOVERNANCE_CACHE_TTL_SECONDS)

        # Permission tables, read from the rule base on first use. None means
        # "not loaded yet", which is distinct from an empty table.
        self._agent_permissions: dict[str, set[str]] | None = None
        self._topic_owners: dict[str, str] | None = None
        self._queue_owners: dict[str, str] | None = None

        # Registered recipients per agent. Deliberately NOT pre-seeded from
        # AGENT_PERMISSIONS any more: reading that table is now a Prolog query,
        # and seeding here would consult the rule base inside __init__ —
        # breaking the laziness this constructor's own docstring promises to
        # the many callers that never issue a governance query. The seed was
        # cosmetic in any case; `register_recipient` already creates a missing
        # sender's set, and every read goes through `.get(sender, set())`.
        self._registered_recipients: dict[str, set[str]] = {}

        # Real Prolog engine (janus-swi) for _query_prolog — lazy, thread-safe.
        self._prolog_engine: Any = prolog_engine
        self._prolog_rules_loaded: bool = False
        self._prolog_lock: threading.Lock = threading.Lock()
        self._prolog_rules_path: str = prolog_rules_path or os.path.join(
            PROLOG_RULES_DIR, _POSTMAN_RULES_FILENAME
        )

    def _ensure_prolog_ready(self) -> Any:
        """Lazily create + consult the Prolog rule base, once, thread-safe.

        Returns:
            The ready ``PrologEngine`` (or injected equivalent).

        Raises:
            Exception: Propagates engine-construction or consult failures
                to the caller (``_evaluate_prolog_goal``), which fails
                CLOSED on any error.
        """
        if self._prolog_engine is not None and self._prolog_rules_loaded:
            return self._prolog_engine
        with self._prolog_lock:
            if self._prolog_engine is None:
                self._prolog_engine = PrologEngine()
            if not self._prolog_rules_loaded:
                self._prolog_engine.consult(self._prolog_rules_path)
                self._prolog_rules_loaded = True
        return self._prolog_engine

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-011"])
    async def check_send_authority(
        self, sender: str, recipient: str, message_type: MessageType
    ) -> AuthResult:
        """Check if sender has authority to send message to recipient.

        Args:
            sender: Agent identifier sending the message
            recipient: Agent identifier receiving the message
            message_type: Type of message being sent

        Returns:
            AuthResult indicating whether operation is authorized
        """
        ctx = {
            "operation": "send",
            "sender": sender,
            "recipient": recipient,
            "message_type": message_type.value,
        }

        if sender == "U:Ox":
            return await self._make_decision(
                True, "Orchestrator has universal send authority", ctx
            )

        if sender not in self.AGENT_PERMISSIONS:
            return await self._make_decision(
                False, f"Unknown sender: {sender}", ctx, escalation=["U:Ox"]
            )

        if recipient in self._registered_recipients.get(sender, set()):
            return await self._make_decision(
                True, f"{recipient} is registered recipient for {sender}", ctx
            )

        permission = f"send.{message_type.value.lower()}"
        if self._has_permission(sender, permission):
            return await self._make_decision(
                True, f"{sender} has permission: {permission}", ctx
            )

        escalation_path = await self._get_escalation_path(sender, "send")
        return await self._make_decision(
            False,
            f"{sender} lacks permission to send {message_type.value} to {recipient}",
            ctx,
            escalation=escalation_path,
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-013"])
    async def check_topic_authority(
        self, sender: str, topic: str, operation: str
    ) -> AuthResult:
        """Check if sender has authority for topic operation.

        Args:
            sender: Agent identifier performing operation
            topic: Topic name
            operation: Operation type (publish, subscribe, create, delete)

        Returns:
            AuthResult indicating whether operation is authorized
        """
        ctx = {"operation": f"topic.{operation}", "sender": sender, "topic": topic}

        if sender == "U:Ox":
            return await self._make_decision(
                True, "Orchestrator has universal topic authority", ctx
            )

        if sender == "U:PS" and operation in {"create", "delete"}:
            return await self._make_decision(True, "Postman can manage topics", ctx)

        if operation == "publish":
            result = await self._check_publish_authority(sender, topic, ctx)
            if result is not None:
                return result

        if operation == "subscribe":
            return await self._make_decision(
                True, "All agents can subscribe to topics", ctx
            )

        escalation_path = await self._get_escalation_path(sender, f"topic.{operation}")
        return await self._make_decision(
            False,
            f"{sender} lacks authority for {operation} on topic {topic}",
            ctx,
            escalation=escalation_path,
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-013"])
    async def _check_publish_authority(
        self, sender: str, topic: str, ctx: dict
    ) -> AuthResult | None:
        """Check publish authority for a topic. Returns AuthResult if authorized, None otherwise."""
        owner = self.TOPIC_OWNERS.get(topic)
        if owner == sender:
            return await self._make_decision(True, f"{sender} owns topic: {topic}", ctx)
        permission = f"send.{topic}"
        if self._has_permission(sender, permission):
            return await self._make_decision(
                True, f"{sender} has permission: {permission}", ctx
            )
        return None

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-013"])
    async def check_queue_authority(
        self, sender: str, queue: str, operation: str
    ) -> AuthResult:
        """Check if sender has authority for queue operation.

        Args:
            sender: Agent identifier performing operation
            queue: Queue name
            operation: Operation type (enqueue, dequeue, create, delete, purge)

        Returns:
            AuthResult indicating whether operation is authorized
        """
        ctx = {"operation": f"queue.{operation}", "sender": sender, "queue": queue}

        if sender == "U:Ox":
            return await self._make_decision(
                True, "Orchestrator has universal queue authority", ctx
            )

        if sender == "U:PS":
            return await self._make_decision(
                True, "Postman has universal queue authority", ctx
            )

        if operation in {"enqueue", "create", "purge", "delete"}:
            owner = self.QUEUE_OWNERS.get(queue)
            if owner == sender:
                return await self._make_decision(
                    True, f"{sender} owns queue: {queue}", ctx
                )

        if operation == "dequeue":
            owner = self.QUEUE_OWNERS.get(queue)
            if owner == sender:
                return await self._make_decision(
                    True, f"{sender} can dequeue from own queue: {queue}", ctx
                )

        escalation_path = await self._get_escalation_path(sender, f"queue.{operation}")
        return await self._make_decision(
            False,
            f"{sender} lacks authority for {operation} on queue {queue}",
            ctx,
            escalation=escalation_path,
        )

    @traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-015"])
    async def check_admin_authority(self, sender: str, operation: str) -> AuthResult:
        """Check if sender has authority for admin operation.

        Args:
            sender: Agent identifier performing operation
            operation: Admin operation type

        Returns:
            AuthResult indicating whether operation is authorized
        """
        ctx = {"operation": f"admin.{operation}", "sender": sender}

        if sender == "U:Ox":
            return await self._make_decision(
                True, "Orchestrator has admin authority", ctx
            )

        if sender == "U:PS" and operation in {"queue", "topic"}:
            return await self._make_decision(
                True, f"Postman has admin.{operation} authority", ctx
            )

        return await self._make_decision(
            False,
            f"{sender} lacks admin authority for {operation}",
            ctx,
            escalation=["U:Ox"],
        )

    async def _make_decision(
        self,
        authorized: bool,
        reason: str,
        context: dict,
        escalation: list[str] | None = None,
    ) -> AuthResult:
        """Build, log, and return an AuthResult.

        Args:
            authorized: Whether the operation is authorized
            reason: Human-readable explanation
            context: Audit log context dict
            escalation: Optional escalation path for denied decisions

        Returns:
            Logged AuthResult
        """
        decision = AuthResult(
            authorized=authorized,
            reason=reason,
            escalation_path=escalation,
        )
        # AUDIT-ONLY Prolog cross-check (SPEC-002): evaluate the real rule
        # base alongside the Python permission tables and record agreement in
        # the audit trail — same staged-rollout posture as the kernel
        # dispatcher's policy audit. Never overrides the Python decision;
        # enforcement is a recorded product decision.
        goal = self._prolog_audit_goal(context)
        if goal is not None:
            predicate, args = goal
            prolog_verdict = await self._query_prolog(predicate, args)
            context = dict(context)
            context["prolog_verdict"] = prolog_verdict
            context["prolog_agrees"] = prolog_verdict == authorized
            if prolog_verdict != authorized:
                logger.warning(
                    "Prolog governance divergence (audit-only): python=%s "
                    "prolog=%s goal=%s(%s) reason=%r",
                    authorized,
                    prolog_verdict,
                    predicate,
                    args,
                    reason,
                )
        decision.audit_id = await self._log_decision(decision, context)
        return decision

    @staticmethod
    def _prolog_audit_goal(context: dict) -> tuple[str, list[Any]] | None:
        """Map a check's audit context to its rule-base goal, if one exists.

        Returns ``(predicate, ground_args)`` for the four authority families,
        or None for contexts with no Prolog-side equivalent.
        """
        operation = str(context.get("operation", ""))
        sender = context.get("sender")
        if sender is None:
            return None
        if operation == "send" and "recipient" in context:
            return (
                "send_authority",
                [
                    sender,
                    context["recipient"],
                    str(context.get("message_type", "")).lower(),
                ],
            )
        if operation.startswith("topic.") and "topic" in context:
            return (
                "topic_authority",
                [sender, context["topic"], operation.split(".", 1)[1]],
            )
        if operation.startswith("queue.") and "queue" in context:
            return (
                "queue_authority",
                [sender, context["queue"], operation.split(".", 1)[1]],
            )
        if operation.startswith("admin."):
            return ("admin_authority", [sender, operation.split(".", 1)[1]])
        return None

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-011"])
    def _has_permission(self, agent: str, permission: str) -> bool:
        """Check if agent has specific permission.

        Args:
            agent: Agent identifier
            permission: Permission string (e.g., "send.code")

        Returns:
            True if agent has permission
        """
        permissions = self.AGENT_PERMISSIONS.get(agent, set())

        # Check for wildcard permission
        if "*" in permissions:
            return True

        # Check for exact permission match
        if permission in permissions:
            return True

        # Check for wildcard category match (e.g., "send.*")
        category = permission.split(".")[0] + ".*"
        if category in permissions:
            return True

        return False

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-005"])
    async def _query_prolog(self, predicate: str, args: list[Any]) -> bool:
        """Query the real postman governance Prolog rule base.

        Consults ``prolog/postman_governance.pl`` (lazily, once, thread-safe)
        via the shared janus-swi ``PrologEngine`` adapter — the same reused
        engine class ``governance.prolog_gate.PrologGate`` consults its rules
        with — then evaluates the ground goal ``predicate(args...)``.

        On ANY failure (janus unavailable, consult error, unknown predicate,
        query error) this fails CLOSED and logs — it never silently allows.

        Args:
            predicate: Prolog predicate name
            args: Ground predicate arguments

        Returns:
            True if the query succeeds at least once, False otherwise
            (including on any engine failure — fail-closed).
        """
        # Create cache key
        cache_key = f"{predicate}({','.join(str(arg) for arg in args)})"

        # Check cache
        if cache_key in self._rule_cache:
            result, timestamp = self._rule_cache[cache_key]
            if (datetime.now(UTC).timestamp() - timestamp) < self._cache_ttl:
                return result

        result = self._evaluate_prolog_goal(predicate, args)

        # Cache result
        self._rule_cache[cache_key] = (result, datetime.now(UTC).timestamp())

        return result

    def _evaluate_prolog_goal(self, predicate: str, args: list[Any]) -> bool:
        """Evaluate a single ground Prolog goal, failing CLOSED on any error.

        Uses ``engine.query()`` (a solution LIST) rather than ``query_once``
        because janus's ``query_once`` returns ``{}`` (not None) for BOTH a
        failed goal and a goal that succeeds with no variable bindings —
        the two are indistinguishable via ``query_once``. ``query()``
        correctly returns ``[]`` on failure and ``[{}]`` (or more) on
        success, matching the pattern already used by
        ``governance.prolog_gate.PrologGate.may_execute``.
        """
        try:
            engine = self._ensure_prolog_ready()
        except Exception as exc:
            logger.error(
                "Prolog engine unavailable for predicate %s (fail-closed deny): %s",
                predicate,
                exc,
            )
            return False

        goal = self._build_prolog_goal(predicate, args)
        try:
            results = self._query_authority(engine, predicate, args, goal)
            return len(results) > 0
        except Exception as exc:
            logger.error("Prolog query %r failed (fail-closed deny): %s", goal, exc)
            return False

    @staticmethod
    def _query_authority(
        engine: Any, predicate: str, args: list[Any], generic_goal: str
    ) -> list[Any]:
        """Issue the goal, spelling the four authority families out LITERALLY.

        Behaviourally identical to ``engine.query(generic_goal)`` — the
        f-strings below build the same text ``_build_prolog_goal`` does, from
        the same ``_to_prolog_term`` rendering. What changes is who can SEE
        them.

        ``scripts/check_prolog_boundary.py`` proves every Python query site
        targets a predicate/arity that exists, by walking the AST for literal
        goal text. A goal assembled at runtime from a ``(predicate, args)``
        pair is invisible to that walk, so this module's entire authority
        surface — ``send_authority/3``, ``topic_authority/3``,
        ``queue_authority/3``, ``admin_authority/2`` and the
        ``permission_matches/3`` they reach — was both unverified by the gate
        and reported by its advisory as "never queried by any discovered
        site". Measured 2026-08-21: all four are queried on EVERY governance
        decision through ``_make_decision``; none of them was visible.

        The generic tail stays for predicates outside the four families
        (``TestQueryPrologReal`` exercises an undefined one to prove the
        fail-closed path), and is deliberately the only non-literal query
        left here.
        """
        # Every `engine.query(...)` below keeps its goal as a LITERAL f-string
        # AT the call site, for the reason the docstring gives. The result is
        # bound to an annotated local before it is returned, rather than
        # returned directly: `engine` is `Any`, so a bare
        # `return engine.query(...)` leaks Any out of a function declared
        # `-> list[Any]` -- mypy's `no-any-return`, and 5 of the 14 errors this
        # file carried on 2026-08-23.
        #
        # The binding states a contract that is already documented and already
        # relied upon, rather than inventing one: `PrologEngine.query` returns
        # a solution LIST (see `_evaluate_prolog_goal`, which calls
        # `len(results)` on it -- an iterator would raise there).
        #
        # The two obvious alternatives are both wrong here. `list(...)` would
        # materialise a copy to satisfy a checker, and hoisting the text into a
        # `goal` variable so there is a single return would make every goal a
        # NON-LITERAL query site -- invisible to `check_prolog_boundary.py`'s
        # AST walk, which is precisely the blindness this method was written to
        # remove.
        solutions: list[Any]
        if predicate == "send_authority" and len(args) == 3:
            sender, recipient, message_type = (_to_prolog_term(a) for a in args)
            solutions = engine.query(
                f"send_authority({sender}, {recipient}, {message_type})"
            )
            return solutions
        if predicate == "topic_authority" and len(args) == 3:
            sender, topic, operation = (_to_prolog_term(a) for a in args)
            solutions = engine.query(f"topic_authority({sender}, {topic}, {operation})")
            return solutions
        if predicate == "queue_authority" and len(args) == 3:
            sender, queue, operation = (_to_prolog_term(a) for a in args)
            solutions = engine.query(f"queue_authority({sender}, {queue}, {operation})")
            return solutions
        if predicate == "admin_authority" and len(args) == 2:
            sender, operation = (_to_prolog_term(a) for a in args)
            solutions = engine.query(f"admin_authority({sender}, {operation})")
            return solutions
        solutions = engine.query(generic_goal)
        return solutions

    @staticmethod
    def _build_prolog_goal(predicate: str, args: list[Any]) -> str:
        """Build a ground Prolog goal string for ``predicate(args...)``."""
        if not args:
            return predicate
        term_args = ",".join(_to_prolog_term(arg) for arg in args)
        return f"{predicate}({term_args})"

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-009"])
    async def _log_decision(self, decision: AuthResult, context: dict) -> str:
        """Log governance decision to audit trail.

        Args:
            decision: Authorization decision result
            context: Additional context for audit log

        Returns:
            Audit log entry identifier
        """
        audit_id = str(uuid.uuid4())
        self._audit_log[audit_id] = {
            "audit_id": audit_id,
            "timestamp": datetime.now(UTC).isoformat(),
            "authorized": decision.authorized,
            "reason": decision.reason,
            "escalation_path": decision.escalation_path,
            **context,
        }
        return audit_id

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-023"])
    async def _get_escalation_path(
        self, sender: str, operation: str
    ) -> list[str] | None:
        """Get escalation path for failed authority check.

        Args:
            sender: Agent identifier
            operation: Operation that failed authorization

        Returns:
            Ordered list of agents who can approve escalation
        """
        # Default escalation path goes through Orchestrator
        return ["U:Ox"]

    def register_recipient(self, sender: str, recipient: str) -> None:
        """Register a recipient for an agent.

        Args:
            sender: Agent identifier
            recipient: Recipient agent identifier
        """
        if sender not in self._registered_recipients:
            self._registered_recipients[sender] = set()
        self._registered_recipients[sender].add(recipient)

    def unregister_recipient(self, sender: str, recipient: str) -> None:
        """Unregister a recipient for an agent.

        Args:
            sender: Agent identifier
            recipient: Recipient agent identifier
        """
        if sender in self._registered_recipients:
            self._registered_recipients[sender].discard(recipient)

    def get_audit_log(self, audit_id: str) -> dict[str, Any] | None:
        """Retrieve audit log entry by ID.

        Args:
            audit_id: Audit log entry identifier

        Returns:
            Audit log entry or None if not found
        """
        return self._audit_log.get(audit_id)

    def get_all_audit_logs(self) -> dict[str, dict[str, Any]]:
        """Retrieve all audit log entries.

        Returns:
            Dictionary of all audit log entries
        """
        return self._audit_log.copy()