orchestrata_ultimate.tools.approval

Reference › orchestrata_ultimate › tools › approval

This module is the governance-backed implementation of the ApprovalGate protocol from registry. It lets the governance rulebase decide whether a volk may run a tool that requires approval. It is imported by agents.tool_registry and tested in tests/tools/test_approval.py and tests/tools/test_approval_coverage.py.

PrologApprovalGate is built with the calling volk's alias as the actor, plus an optional client for tests. It satisfies the protocol structurally, with no inheritance, so the tools package keeps no hard governance import. approve asks the governance client whether may_execute_tool/2 holds for the tool name and actor. The call arguments are accepted but not consulted, so a grant is per volk and tool and never depends on arguments. An unattributed call (actor None) is denied without querying. If the query raises, the gate logs a warning and denies, so an unreachable rulebase never approves. _governance_client builds the shared client lazily on first use and keeps it for the process.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PrologApprovalGate

Line 56

Grants an approval-requiring tool iff may_execute_tool/2 says so.

Functions

Source

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

"""The governance-Prolog implementation of :class:`ApprovalGate`.

:meth:`ToolRegistry.execute` refuses a tool whose metadata sets
``requires_approval`` unless an ``ApprovalGate`` grants it, and refuses
outright when no gate is installed. This is the gate production installs.

The decision lives in ``governance.pl`` as ``may_execute_tool/2`` facts, so
the human grants a tool to a volk once, in a rulebase, rather than answering
per call. Absence of a fact is a denial — the predicate has no default
clause.

The Prolog engine is process-global (``get_prolog_worker``), so the bridge and
client are built once, lazily, behind a lock, and every per-volk gate shares
them. A gate instance is just an actor alias plus that shared client.
"""

from __future__ import annotations

import threading
from collections.abc import Mapping
from typing import Any

from orchestrata_ultimate.core.logging import get_logger

_logger = get_logger(__name__)

_client_lock = threading.Lock()
_shared_client: Any | None = None


def _governance_client() -> Any:
    """Build (once) the governance client every approval gate queries through.

    Imported lazily inside the function: ``tools`` must not take a hard import
    dependency on ``governance``/janus just to define the gate type, and the
    Prolog engine should not be spun up for a process that never executes an
    approval-requiring tool.
    """
    global _shared_client
    if _shared_client is not None:
        return _shared_client
    with _client_lock:
        if _shared_client is None:
            from orchestrata_ultimate.governance.prolog_client import (
                GovernancePrologClient,
                load_governance_rules,
            )
            from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

            bridge = PrologBridge()
            load_governance_rules(bridge)
            _shared_client = GovernancePrologClient(bridge)
    return _shared_client


class PrologApprovalGate:
    """Grants an approval-requiring tool iff ``may_execute_tool/2`` says so.

    Satisfies the :class:`~orchestrata_ultimate.tools.registry.ApprovalGate`
    protocol structurally (no inheritance), so ``tools`` keeps its
    ``governance``-free import surface.

    *actor* is the calling volk's alias. ``None`` — a tool call that arrives
    unattributed — is denied without querying: an unattributed request cannot
    be attributed to a granted volk.
    """

    def __init__(self, actor: str | None, client: Any | None = None) -> None:
        self._actor = actor
        # `client` is the injection seam for tests, which supply a client over
        # a bridge holding a scratch rulebase. Production passes nothing and
        # shares the process-wide engine.
        self._client = client

    def approve(self, tool_name: str, arguments: Mapping[str, Any]) -> bool:
        """Return True to let the call proceed, False to refuse it.

        *arguments* is accepted to satisfy the protocol and is deliberately
        NOT consulted: the grant is per (volk, tool), so an argument-dependent
        decision would make the same tool sometimes-approved in a way no fact
        in the rulebase records.
        """
        client = self._client if self._client is not None else _governance_client()
        try:
            return bool(client.may_execute_tool(tool_name, self._actor))
        except Exception as exc:
            # Fail closed: an approval gate that cannot reach its rulebase
            # must not become a rubber stamp.
            _logger.warning(
                "Approval query for %s (actor=%s) failed: %s — denying",
                tool_name,
                self._actor,
                exc,
            )
            return False