orchestrata_ultimate.governance.prolog_client

Reference › orchestrata_ultimate › governance › prolog_client

The governance-flavoured Prolog client, and the largest module of the package. GovernancePrologClient wraps the shared PrologBridge (janus-swi, from core.prolog_adapter) and turns Python questions such as "may this actor execute this action" into Prolog goals, then turns the answers back into booleans. policy_engine, constitutional, ethical, resource_access, safe_mode, dynamic_policy and the gateway server use it. It serves SPEC-007 (REQ-007-FR10-032, REQ-007-US-001-001, -007 and the constitutional, resource and safe-mode requirements).

Loading rules: load_governance_rules(bridge, rules_dir) is described in its docstring as the one blessed loader. It consults every file in GOVERNANCE_RULE_FILES into the bridge and returns the number loaded, logging a warning for each file that is missing or fails to consult. dispatcher and cli_main call it at boot; the docstring records that an earlier hand-picked subset left ownership checks unable to discriminate. count_governance_predicates counts what was loaded for the boot log line.

The query methods, each mapping to a Prolog predicate:

• may_execute(action, actor, context) asks governance:may_execute/2 and returns the boolean; may_execute_explained returns a ProofResult; may_execute_tool is the tool-level variant.
• constitutional_check(article, action, context) asks constitutional_check/3.
• has_permission and may_access (dispatched by resource type to the file, network, API or database predicate) answer resource questions; an unrecognised resource type is resolved by the strict flag, never treated as a file.
• ethical_check and valid_policy cover ethics and rule validation.
• track_violation, enter_safe_mode, exit_safe_mode and is_safe_mode assert and read the violation and safe-mode facts; assert_volk_operational, retract_volk_operational and set_volk_active use the goals from volk_facts.

Failure handling is uniform. A query that raises or returns an error goes through governance_strict (see _strict): strict mode, the default, denies, and non-strict mode allows or falls back to evaluate_policy. Every answer is logged through make_verdict from verdict as a policy or error verdict. Atoms are quoted by _format_atom before they go into a goal.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

GovernancePrologClient

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

Governance-specific wrapper around PrologBridge.

Functions

Source

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

"""Governance-specific Prolog client for Orchestrata.

Wraps PrologBridge with governance-domain operations that map
to Prolog predicates in governance.pl, constitutional.pl, ethics.pl,
and resources.pl.
"""

from __future__ import annotations

import hashlib
import logging
import re
from collections.abc import Mapping
from pathlib import Path
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from orchestrata_ultimate.core.prolog_adapter import ProofResult
    from orchestrata_ultimate.infra.prolog_bridge import (
        PrologBridge,
        QueryResult,
    )

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.enums import FactScope
from orchestrata_ultimate.core.infra_config import PROLOG_RULES_DIR
from orchestrata_ultimate.governance._strict import governance_strict
from orchestrata_ultimate.governance.exceptions import PrologClientError
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__)

# The governance-domain Prolog rule files that back every predicate
# GovernancePrologClient queries (constitutional_check/3, ethical_harm_check/3,
# has_permission/3, may_access_file/3, may_execute/2, tool_allowed/3,
# evaluate_policy/3, ...). Mirrors the curated list PrologGate.load_rules()
# consults (governance/prolog_gate.py) plus resources.pl, which
# GovernancePrologClient.may_access's fallback and has_permission depend on.
#
# completion.pl is CONSULTED here too (not runtime-asserted) for the same
# reason PrologGate consults it: its variable-bearing clauses only load
# correctly via consult on the janus-swi backend.
#
# W1-P3 audit: intent_policy.pl (SPEC-003 permission_granted/2 --
# IntentValidator's fail-closed terminal predicate) and
# postman_governance.pl (SPEC-002 PostmanGovernanceManager's send/topic/
# queue/admin authority rules) are governance-domain rule files the original
# curated list omitted (proposal `reports/proposal-wire-prolog-governance.md`
# §3 finding F5); neither declares a predicate name that collides with the
# other files, and both are included below. The e2e enforcement-proof tests
# count ``len(GOVERNANCE_RULE_FILES)`` rather than a hard-coded number, so
# the list can grow without silently breaking them.
GOVERNANCE_RULE_FILES: tuple[str, ...] = (
    "governance.pl",
    "constitutional.pl",
    "agents.pl",
    "ethics.pl",
    "pam_rules.pl",
    "resources.pl",
    "completion.pl",
    # Governance-domain files identified by the W1-P3 audit (proposal §3
    # F5): the SPEC-003 intent policy (default-allow rule base the intent
    # validator queries) and the postman governance predicates. Landed
    # together with the e2e assertions that count this tuple's length.
    "intent_policy.pl",
    "postman_governance.pl",
)


def load_governance_rules(
    bridge: PrologBridge, rules_dir: str | Path | None = None
) -> int:
    """Consult the COMPLETE governance rule corpus (``GOVERNANCE_RULE_FILES``)
    into ``bridge``'s Prolog engine.

    This is the ONE blessed loader (W1-P3): every production construction
    site that needs governance verdicts calls this instead of hand-picking
    its own rule-file subset. It is the wiring step that was missing at the
    original production call site (gateway/server.py's ``_make_policy_engine``
    factory): without it, ``constitutional_check/3``, ``ethical_harm_check/3``,
    ``has_permission/3``, etc. are undefined predicates in the janus-swi
    engine, so every GovernancePrologClient/EthicalBoundary/
    ConstitutionalEnforcer query hits its except-block safe default and
    enforcement is a silent no-op (SPEC-007). A second production site,
    ``kernel/dispatcher.py::_build_default_policy_evaluator``, hand-loaded
    only ``governance.pl`` and so never consulted ``pam_rules.pl`` --
    ``governance.pl``'s ``check_constraint/4`` (ownership) delegates to
    ``pam:pam_owns/2``, which errored into ``fail`` and made
    ``PolicyEvaluator.validate_ownership`` silently unable to discriminate
    ownership in production; it now calls this loader too.

    Idempotent: ``PrologBridge.load_rules`` tracks consulted paths per
    instance and skips re-consulting a file already loaded on that bridge,
    so calling this twice on the same bridge (or passing a bridge some
    files were already consulted into) is a safe no-op for those files.

    Args:
        bridge: The PrologBridge whose engine should have the rules consulted.
        rules_dir: Directory containing the .pl files. Defaults to
            ``PROLOG_RULES_DIR`` (src/orchestrata_ultimate/prolog/).

    Returns:
        Number of rule files successfully consulted.
    """
    directory = Path(rules_dir) if rules_dir is not None else Path(PROLOG_RULES_DIR)
    loaded = 0
    for filename in GOVERNANCE_RULE_FILES:
        if bridge.load_rules(directory / filename):
            loaded += 1
        else:
            _logger.warning(
                "Governance rule file not loaded (missing or consult failed): %s",
                directory / filename,
            )
    return loaded


_MODULE_DECL_RE = re.compile(r"^:-\s*module\((\w+)\s*,", re.MULTILINE)


def _governance_modules(rules_dir: str | Path) -> set[str]:
    """Module names declared by the files in ``GOVERNANCE_RULE_FILES``, plus
    ``user`` -- ``completion.pl`` and ``intent_policy.pl`` ship no ``:-
    module(...)`` declaration, so their clauses consult into ``user``."""
    modules = {"user"}
    for filename in GOVERNANCE_RULE_FILES:
        path = Path(rules_dir) / filename
        try:
            text = path.read_text(encoding="utf-8")
        except OSError:
            continue
        match = _MODULE_DECL_RE.search(text)
        if match:
            modules.add(match.group(1))
    return modules


def count_governance_predicates(
    bridge: PrologBridge, rules_dir: str | Path | None = None
) -> int:
    """Count the distinct predicate NAMES ``current_predicate/1`` reports
    across the governance-domain modules (W1-P6 boot health line).

    This is the "did the rules actually load" signal for the boot log: a
    rule base that consulted zero files reports 0 here too (no module has
    any predicate), making that failure mode unmistakable rather than a
    silent no-op (the exact class of incidents 1-3 in
    ``reports/proposal-wire-prolog-governance.md``).
    """
    directory = Path(rules_dir) if rules_dir is not None else Path(PROLOG_RULES_DIR)
    names: set[str] = set()
    for module in sorted(_governance_modules(directory)):
        result = bridge.query(f"current_predicate({module}:N/_)")
        for sol in result.solutions:
            value = sol.get("N")
            if isinstance(value, str):
                names.add(value)
    return len(names)


@traces(
    spec="000",
    story="US-007",
    reqs=["REQ-000-066"],
)
class GovernancePrologClient:
    """Governance-specific wrapper around PrologBridge.

    Maps governance operations to Prolog queries and fact assertions.
    All 10 core governance predicates from the spec are exposed here.
    """

    def __init__(
        self, bridge: PrologBridge, env: Mapping[str, str] | None = None
    ) -> None:
        self._bridge = bridge
        self._env = env

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

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

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-032",
            "REQ-007-FR10-034",
            "REQ-007-US-001-007",
        ],
    )
    def _query_bool_safe(self, query_str: str) -> bool:
        """Execute a boolean query with error handling."""
        try:
            return self._bridge.query_bool(query_str)
        except Exception as exc:
            _logger.error("Prolog query failed: %s — %s", query_str, exc)
            raise PrologClientError(f"Query failed: {query_str}") from exc

    @traces(spec="007", story="FR-PGOV-010", reqs=["REQ-007-FR10-034"])
    def _query_safe(self, query_str: str) -> QueryResult:
        """Execute a query with error handling."""
        try:
            return self._bridge.query(query_str)
        except Exception as exc:
            _logger.error("Prolog query failed: %s — %s", query_str, exc)
            raise PrologClientError(f"Query failed: {query_str}") from exc

    # ------------------------------------------------------------------
    # 1. may_execute/3
    # ------------------------------------------------------------------
    def may_execute_tool(self, tool_name: str, actor: str | None) -> bool:
        """Whether *actor* may execute the approval-requiring tool *tool_name*.

        Maps to ``governance:may_execute_tool/2`` — a STRICT allowlist with no
        default clause, so an unmatched pair fails and failure means denied.

        Deliberately not :meth:`may_execute`. That predicate is default-ALLOW
        (its second clause succeeds when the agent is unregistered, and
        ``agent_action_blocked/2`` carries no facts), so it returns true for
        every pair and cannot carry an approval decision.

        Fails closed on every uncertainty: a caller with no alias is denied
        without querying at all, since an unattributed request cannot be
        attributed to a granted volk; and a query that ERRORS — rules not
        consulted, engine unavailable — is denied rather than allowed, which
        is the opposite of :meth:`may_execute`'s permissive fallback.
        """
        if not actor:
            make_verdict(
                surface="GovernancePrologClient.may_execute_tool",
                decision="deny",
                is_deny=True,
                is_error=False,
                rule_source="governance:may_execute_tool/2",
                reason=(
                    f"{tool_name} requires approval and the call carried no "
                    f"actor alias — an unattributed request cannot be granted"
                ),
                logger_=_logger,
            )
            return False

        query_str = (
            f"governance:may_execute_tool("
            f"{self._format_atom(actor)}, {self._format_atom(f'tool_{tool_name}')})"
        )
        try:
            result = self._bridge.query(query_str)
        except Exception as exc:  # pragma: no cover - defensive
            make_verdict(
                surface="GovernancePrologClient.may_execute_tool",
                decision="deny",
                is_deny=True,
                is_error=True,
                rule_source="governance:may_execute_tool/2",
                reason=f"query raised {exc} — failing closed",
                logger_=_logger,
            )
            return False
        if result.error is not None:
            make_verdict(
                surface="GovernancePrologClient.may_execute_tool",
                decision="deny",
                is_deny=True,
                is_error=True,
                rule_source="governance:may_execute_tool/2",
                reason=f"query errored ({result.error}) — failing closed",
                logger_=_logger,
            )
            return False

        allowed = result.success
        make_verdict(
            surface="GovernancePrologClient.may_execute_tool",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=False,
            rule_source="governance:may_execute_tool/2",
            reason=(
                f"may_execute_tool({actor}, tool_{tool_name}) matched a fact"
                if allowed
                else f"no may_execute_tool fact grants tool_{tool_name} to {actor}"
            ),
            logger_=_logger,
        )
        return allowed

    # ------------------------------------------------------------------
    # Volk fact sync -- what makes may_execute/2 able to say no
    # ------------------------------------------------------------------
    #
    # The facts themselves, and why they are these facts, live in
    # ``governance.volk_facts``: PrologGate runs the identical goals against a
    # different engine, and two copies of the list is how the two gates start
    # disagreeing about what "operational" means. This class supplies only the
    # execution.
    #
    # All three are best-effort by contract: registration must not fail
    # because the rule base is unavailable. False means the volk stays UNKNOWN
    # to governance, which under the deny-unknown posture means denied — the
    # safe direction to fail in.

    def _run_goals(self, goals: tuple[str, ...], what: str, alias: str) -> bool:
        try:
            for goal in goals:
                self._bridge.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_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_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_goals(active_goals(alias, active), "volk_active update", alias)

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-032",
            "REQ-007-US-001-001",
            "REQ-007-US-001-007",
        ],
    )
    def may_execute(self, action: str, actor: str, context: dict[str, Any]) -> bool:
        """Check if an actor may execute an action in a given context.

        Maps to Prolog: governance:may_execute(Agent, Action) — the /2
        predicate governance.pl actually declares (SPEC-007 FR-GOV-SPIKE-1;
        REQ-007-US-001-001). The agent comes FIRST. The module qualifier makes
        resolution independent of whether the export was imported into
        ``user`` (janus state is process-global; import order varies).
        ``context`` feeds only the evaluate_policy/3 fallback, which runs if
        the query itself errors (e.g. rules not consulted) and strict mode
        is off.
        """
        action_atom = self._format_atom(action)
        actor_atom = self._format_atom(actor)
        query_str = f"governance:may_execute({actor_atom}, {action_atom})"
        try:
            # W1-P6: query() (not query_bool()) so the ``error`` field is
            # available -- it is the ONLY reliable POLICY-vs-ERROR signal
            # (see governance/verdict.py's module docstring): PrologBridge.
            # query already swallows every exception internally into
            # QueryResult(success=False, error=<msg>), so a bare bool loses
            # exactly the distinction this method needs to attribute.
            result = self._bridge.query(query_str)
        except Exception as exc:
            return self._may_execute_error_verdict(str(exc), action, actor, context)
        if result.error is not None:
            return self._may_execute_error_verdict(result.error, action, actor, context)

        allowed = result.success
        explanation = (
            None
            if allowed
            else explain_may_execute_deny(self._bridge.query_bool, actor, action)
        )
        reason = (
            "governance:may_execute/2 matched a clause"
            if allowed
            else "governance:may_execute/2 had no matching clause"
        )
        if explanation:
            reason = f"{reason} — {explanation}"
        make_verdict(
            surface="GovernancePrologClient.may_execute",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=False,
            rule_source="governance:may_execute/2",
            reason=reason,
            logger_=_logger,
        )
        return allowed

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-032",
            "REQ-007-US-001-001",
            "REQ-007-US-001-007",
        ],
    )
    def _may_execute_error_verdict(
        self, error: str, action: str, actor: str, context: dict[str, Any]
    ) -> bool:
        """The query itself errored (undefined predicate, janus exception,
        or a genuinely broken bridge). Resolve via governance_strict() like
        every other governance error branch, then label the verdict ERROR
        (W1-P6) so it is never confused with a real policy verdict."""
        _logger.error("may_execute Prolog query failed: %s", error)
        if governance_strict(self._env):
            allowed = False
        else:
            # Fallback: use evaluate_policy/3 which exists in governance.pl.
            # Deliberately unguarded, matching the pre-P6 behavior: if the
            # bridge is genuinely broken (e.g. None), this raises and
            # propagates -- it is not this method's job to mask that.
            allowed = self._evaluate_via_policy(action, context)
        make_verdict(
            surface="GovernancePrologClient.may_execute",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=True,
            rule_source="error:governance:may_execute/2",
            reason=f"query error: {error}",
            logger_=_logger,
        )
        return allowed

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-032",
            "REQ-007-US-001-001",
            "REQ-007-US-001-007",
        ],
    )
    def may_execute_explained(
        self, action: str, actor: str, context: dict[str, Any]
    ) -> ProofResult:
        """Opt-in explaining variant of the may_execute decision (ADR-0031).

        Explains the SAME governance-domain rule ``governance:may_execute(Agent,
        Action)`` (agent-first, per governance.pl) that decides whether
        ``actor`` may perform ``action`` -- naming, on success, which rule
        fired and which facts it consumed, and on denial, the first failing
        precondition of every candidate clause. This is deliberately a
        SEPARATE query from ``may_execute`` (never delegates to it): the two
        must stay independently correct against governance.pl, and
        ``may_execute`` itself is frozen -- unchanged by this method, as its
        own tests prove.

        Persistence: this decision is not currently written to any audit
        record on this call path (``may_execute`` has none either).
        # B19: audit persistence lands with the gateway audit writer.
        """
        # Local import (not a module-level one): core.prolog_adapter imports
        # janus_swi at ITS OWN module top level, which eagerly bootstraps the
        # SWI engine (_swipl.initialize). Every other Prolog access on this
        # class already goes through PrologBridge/PrologWorker, which defers
        # that bootstrap to first real use; a top-level import here would
        # silently change GovernancePrologClient's import-time behavior for
        # every caller, not just this method.
        from orchestrata_ultimate.core.prolog_adapter import ProofResult

        actor_atom = self._format_atom(actor)
        action_atom = self._format_atom(action)
        goal = f"governance:may_execute({actor_atom}, {action_atom})"
        try:
            result: ProofResult = self._bridge._prolog.query_explained(goal)
            return result
        except Exception as exc:
            _logger.error("may_execute_explained Prolog query failed: %s", exc)
            return ProofResult(succeeded=False, tree=None, whynot=[], goal=goal)

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=[
            "REQ-007-FR10-032",
            "REQ-007-US-001-001",
            "REQ-007-US-001-007",
        ],
    )
    def _evaluate_via_policy(self, action: str, context: dict[str, Any]) -> bool:
        """Fallback evaluation using existing evaluate_policy/3."""
        action_atom = self._format_atom(action)
        resource = context.get("resource", "unknown")
        resource_atom = self._format_atom(resource)
        result = self._bridge.query(
            f"evaluate_policy({action_atom}, {resource_atom}, Decision)"
        )
        if result.success and result.solutions:
            decision = result.solutions[0].get("Decision", "deny")
            if isinstance(decision, bytes):
                decision = decision.decode()
            return str(decision).lower() in ("allow", "true")
        return False

    # ------------------------------------------------------------------
    # 2. constitutional_check/3
    # ------------------------------------------------------------------
    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=["REQ-007-FR1-003", "REQ-007-US-002-008"],
    )
    def constitutional_check(
        self, article: int, action: str, context: dict[str, Any] | None = None
    ) -> bool:
        """Check if action complies with a constitutional article.

        Maps to Prolog: constitutional_check(Article, Action, Result).
        """
        action_atom = self._format_atom(action)
        session = self._format_atom((context or {}).get("session_id", "unknown"))
        try:
            # ``article`` is embedded directly (not re-embedded below) --
            # constructing a SECOND f-string with ``article`` in the error
            # path would risk re-raising on a caller whose article value
            # only formats successfully once (see
            # test_constitutional_check_fails_closed_on_internal_exception_by_default).
            query_str = f"constitutional_check({article}, {action_atom}, {session})"
            result = self._bridge.query(query_str)
        except Exception as exc:
            _logger.error(
                "constitutional_check/%d Prolog query failed: %s", article, exc
            )
            allowed = not governance_strict(self._env)
            make_verdict(
                surface="GovernancePrologClient.constitutional_check",
                decision="allow" if allowed else "deny",
                is_deny=not allowed,
                is_error=True,
                rule_source="error:constitutional_check/3",
                reason=f"query error: {exc}",
                logger_=_logger,
            )
            return allowed

        if result.error is not None:
            _logger.error(
                "constitutional_check/%d Prolog query failed: %s",
                article,
                result.error,
            )
            allowed = not governance_strict(self._env)
            make_verdict(
                surface="GovernancePrologClient.constitutional_check",
                decision="allow" if allowed else "deny",
                is_deny=not allowed,
                is_error=True,
                rule_source="error:constitutional_check/3",
                reason=f"query error: {result.error}",
                logger_=_logger,
            )
            return allowed

        allowed = result.success
        make_verdict(
            surface="GovernancePrologClient.constitutional_check",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=False,
            rule_source="constitutional_check/3",
            reason=(
                "constitutional_check/3 matched a clause"
                if allowed
                else "constitutional_check/3 had no matching clause"
            ),
            logger_=_logger,
        )
        return allowed

    # ------------------------------------------------------------------
    # 3. has_permission/3
    # ------------------------------------------------------------------
    @traces(
        spec="007",
        story="US-PGOV-004",
        reqs=[
            "REQ-007-FR1-004",
            "REQ-007-FR3-001",
            "REQ-007-US-004-008",
        ],
    )
    def has_permission(self, actor: str, action: str, resource: str) -> bool:
        """Check if actor has permission for action on resource.

        Maps to Prolog: has_permission(Volk, Resource, Operation) -- the
        arity/ORDER pam_rules.pl actually declares (F3 fix). The Python
        signature keeps (actor, action, resource) for backward
        compatibility with existing callers; only the query's argument
        ORDER changes, from (Actor, Action, Resource) to (Actor, Resource,
        Action), to match pam_rules.pl:109,113. Before this fix, args 2/3
        were swapped relative to the shipped predicate, so every call
        queried a permutation that could never match a real pam_owns/
        grant_active fact -- has_permission always failed silently.
        """
        query_str = (
            f"has_permission({self._format_atom(actor)}, "
            f"{self._format_atom(resource)}, {self._format_atom(action)})"
        )
        try:
            allowed = self._query_bool_safe(query_str)
        except PrologClientError as exc:
            # _query_bool_safe raises on a genuine internal failure -- keep
            # that contract EXACTLY (no test drives it today, but it is the
            # documented behavior of the shared helper); log+re-raise so the
            # verdict is still attributed before the exception propagates.
            make_verdict(
                surface="GovernancePrologClient.has_permission",
                decision="deny",
                is_deny=True,
                is_error=True,
                rule_source="error:has_permission/3",
                reason=f"query error: {exc}",
                logger_=_logger,
            )
            raise
        make_verdict(
            surface="GovernancePrologClient.has_permission",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=False,
            rule_source="pam:has_permission/3",
            reason=(
                "pam:has_permission/3 matched a clause"
                if allowed
                else "no matching has_permission/3 clause"
            ),
            logger_=_logger,
        )
        return allowed

    # ------------------------------------------------------------------
    # 4. may_access/4
    # ------------------------------------------------------------------
    # F5: resources.pl declares FOUR resource-kind predicates
    # (may_access_file/3, may_access_network/3, may_access_api/3,
    # may_access_db/3), each with a different meaning for its third
    # argument (Mode: read/write/execute/delete; Port: a number; Method:
    # get/post/...; Operation: select/insert/...). may_access's own
    # signature carries no resource-kind parameter, so the caller supplies
    # it via context["resource_type"] (defaulting to "file" -- the ONLY
    # kind the pre-fix code ever queried, so existing callers see no
    # behavior change unless they opt in to a different kind).
    _RESOURCE_ACCESS_PREDICATES: dict[str, str] = {
        "file": "may_access_file",
        "network": "may_access_network",
        "api": "may_access_api",
        "db": "may_access_db",
    }

    @traces(
        spec="007",
        story="US-PGOV-004",
        reqs=[
            "REQ-007-FR1-005",
            "REQ-007-FR3-005",
            "REQ-007-FR4-002",
            "REQ-007-US-004-005",
        ],
    )
    def may_access(
        self,
        actor: str,
        resource: str,
        mode: str,
        context: dict[str, Any] | None = None,
    ) -> bool:
        """Check resource access permission.

        Maps to Prolog: may_access_file/3, may_access_network/3,
        may_access_api/3, or may_access_db/3 -- dispatched by
        ``context["resource_type"]`` (F5 fix; before this fix every mode
        hardcoded may_access_file/3, so may_access_network/api/db were
        unreachable dead code even though resources.pl ships all three).
        An unrecognized resource_type is a genuine "unknown mode" --
        never silently treated as file mode -- and resolves via
        governance_strict() like every other error branch in this class.
        """
        context = context or {}
        resource_type = context.get("resource_type", "file")
        predicate = self._RESOURCE_ACCESS_PREDICATES.get(resource_type)
        actor_atom = self._format_atom(actor)
        resource_atom = self._format_atom(resource)
        if predicate is None:
            _logger.error(
                "may_access: unrecognized resource_type %r for actor=%s resource=%s",
                resource_type,
                actor,
                resource,
            )
            allowed = not governance_strict(self._env)
            make_verdict(
                surface="GovernancePrologClient.may_access",
                decision="allow" if allowed else "deny",
                is_deny=not allowed,
                is_error=True,
                rule_source=f"error:unrecognized_resource_type:{resource_type}",
                reason=f"unrecognized resource_type {resource_type!r}",
                logger_=_logger,
            )
            return allowed
        # may_access_network/3's third argument is a Port NUMBER, not an
        # atom (resources.pl:119-127) -- pass it unquoted when it parses
        # as one so number(Port) can actually succeed; every other kind's
        # third argument is a mode/method/operation atom.
        if resource_type == "network" and mode.isdigit():
            mode_term = mode
        else:
            mode_term = self._format_atom(mode)
        query_str = f"{predicate}({actor_atom}, {resource_atom}, {mode_term})"
        try:
            result = self._bridge.query(query_str)
        except Exception as exc:
            return self._may_access_error_fallback(
                predicate, str(exc), mode, resource_atom, context
            )
        if result.error is not None:
            return self._may_access_error_fallback(
                predicate, result.error, mode, resource_atom, context
            )

        allowed = result.success
        make_verdict(
            surface="GovernancePrologClient.may_access",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=False,
            rule_source=f"{predicate}/3",
            reason=(
                f"{predicate}/3 matched a clause"
                if allowed
                else f"no matching {predicate}/3 clause"
            ),
            logger_=_logger,
        )
        return allowed

    @traces(
        spec="007", story="US-PGOV-004", reqs=["REQ-007-FR4-002", "REQ-007-US-004-005"]
    )
    def _may_access_error_fallback(
        self,
        predicate: str,
        error: str,
        mode: str,
        resource_atom: str,
        context: dict[str, Any],
    ) -> bool:
        """The resource-kind query errored. Deny under strict; otherwise
        fall back to ``tool_allowed/3`` (governance.pl), deliberately
        unguarded like the pre-P6 code -- if the bridge is genuinely broken
        this fallback also raises, and it is not this helper's job to mask
        that."""
        _logger.error("%s Prolog query failed: %s", predicate, error)
        if governance_strict(self._env):
            make_verdict(
                surface="GovernancePrologClient.may_access",
                decision="deny",
                is_deny=True,
                is_error=True,
                rule_source=f"error:{predicate}/3",
                reason=f"query error: {error} — fail-closed",
                logger_=_logger,
            )
            return False
        # Fallback: use tool_allowed/3 which exists in governance.pl.
        # Intent must be the intent(Action, Resource, Constraints) term
        # tool_allowed/3's own body (policy_permits/2, intent_action/2)
        # expects -- a bare resource atom there would never match.
        tool = "read" if mode == "read" else "write"
        intent_term = f"intent({tool}, {resource_atom}, [])"
        allowed = self._bridge.query_bool(
            f"tool_allowed({self._format_atom(tool)}, {intent_term}, "
            f"{self._format_atom(context.get('session_id', 'unknown'))})"
        )
        make_verdict(
            surface="GovernancePrologClient.may_access",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=True,
            rule_source=f"error:{predicate}/3 fallback=tool_allowed/3",
            reason=f"query error: {error} — permissive fallback via tool_allowed/3",
            logger_=_logger,
        )
        return allowed

    # ------------------------------------------------------------------
    # 5. ethical_check/2
    # ------------------------------------------------------------------
    @traces(spec="007", story="FR-PGOV-001", reqs=["REQ-007-FR1-006"])
    def ethical_check(self, action: str, context: dict[str, Any] | None = None) -> bool:
        """Check if action passes ethical boundaries.

        Maps to Prolog: ethical_check(Action, Result).
        """
        action_atom = self._format_atom(action)
        query_str = f"ethical_check({action_atom}, pass)"
        try:
            result = self._bridge.query(query_str)
        except Exception as exc:
            _logger.error("ethical_check Prolog query failed: %s", exc)
            allowed = not governance_strict(self._env)
            make_verdict(
                surface="GovernancePrologClient.ethical_check",
                decision="allow" if allowed else "deny",
                is_deny=not allowed,
                is_error=True,
                rule_source="error:ethical_check/2",
                reason=f"query error: {exc}",
                logger_=_logger,
            )
            return allowed

        if result.error is not None:
            _logger.error("ethical_check Prolog query failed: %s", result.error)
            allowed = not governance_strict(self._env)
            make_verdict(
                surface="GovernancePrologClient.ethical_check",
                decision="allow" if allowed else "deny",
                is_deny=not allowed,
                is_error=True,
                rule_source="error:ethical_check/2",
                reason=f"query error: {result.error}",
                logger_=_logger,
            )
            return allowed

        allowed = result.success
        make_verdict(
            surface="GovernancePrologClient.ethical_check",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=False,
            rule_source="ethical_check/2",
            reason=(
                "ethical_check/2 matched a clause"
                if allowed
                else "no matching ethical_check/2 clause"
            ),
            logger_=_logger,
        )
        return allowed

    # ------------------------------------------------------------------
    # 6. valid_policy/1
    # ------------------------------------------------------------------
    @traces(
        spec="007",
        story="US-PGOV-003",
        reqs=[
            "REQ-007-FR6-001",
            "REQ-007-US-003-002",
            "REQ-007-US-003-008",
        ],
    )
    def valid_policy(self, rule: dict[str, Any]) -> bool:
        """Validate a policy rule before activation.

        Maps to Prolog: valid_policy(Decision) -- governance.pl:
        validates that the rule's ``decision`` is one of the recognized
        governance verdicts (F6 fix). Before this fix, valid_policy/1 was
        declared nowhere, and the caller treated the resulting
        existence_error as "no validation rule -> allow" unconditionally
        (allow-by-absence, ignoring strictness entirely) -- absence now
        denies under strict mode like every other error branch here.
        """
        decision = str(rule.get("decision", "allow"))
        decision_atom = self._format_atom(decision)
        query_str = f"valid_policy({decision_atom})"
        try:
            result = self._bridge.query(query_str)
        except Exception as exc:
            _logger.error("valid_policy Prolog query failed: %s", exc)
            allowed = not governance_strict(self._env)
            make_verdict(
                surface="GovernancePrologClient.valid_policy",
                decision="allow" if allowed else "deny",
                is_deny=not allowed,
                is_error=True,
                rule_source="error:valid_policy/1",
                reason=f"query error: {exc}",
                logger_=_logger,
            )
            return allowed

        is_error = result.error is not None
        if is_error:
            _logger.error("valid_policy Prolog query failed: %s", result.error)
        if not result.success:
            # NOTE (W1-P6): this applies the strict-mode fallback even when
            # the query genuinely executed with zero solutions
            # (result.error is None) -- a pre-existing F6 design choice
            # (valid_policy/1's absence and a genuine non-match were
            # historically treated identically; preserved exactly here).
            # ``is_error`` below still keys off result.error, so a TRUE
            # query failure (undefined predicate) stays distinguishable in
            # the log from a real "no" against the loaded rule base.
            allowed = not governance_strict(self._env)
        else:
            allowed = len(result.solutions) > 0
        make_verdict(
            surface="GovernancePrologClient.valid_policy",
            decision="allow" if allowed else "deny",
            is_deny=not allowed,
            is_error=is_error,
            rule_source="error:valid_policy/1" if is_error else "valid_policy/1",
            reason=(
                f"query error: {result.error}"
                if is_error
                else (
                    "valid_policy/1 matched a clause"
                    if allowed
                    else "no matching valid_policy/1 clause"
                )
            ),
            logger_=_logger,
        )
        return allowed

    # ------------------------------------------------------------------
    # 7. track_violation/4
    # ------------------------------------------------------------------
    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=[
            "REQ-007-FR7-001",
            "REQ-007-US-002-004",
        ],
    )
    def track_violation(
        self,
        actor: str,
        action: str,
        violation_type: str,
        details: str,
    ) -> str:
        """Record a governance violation as a Prolog fact.

        Maps to Prolog: assertz(violation(Actor, Action, Type, Details)).
        Returns the fact ID.
        """
        return self._bridge.assert_fact(
            "violation",
            [actor, action, violation_type, details],
            FactScope.SESSION,
        )

    # ------------------------------------------------------------------
    # 8. enter_safe_mode/1
    # ------------------------------------------------------------------
    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=[
            "REQ-007-FR8-005",
            "REQ-007-FR8-008",
            "REQ-007-US-002-010",
        ],
    )
    def enter_safe_mode(self, reason: str) -> str:
        """Assert safe_mode fact in Prolog.

        Maps to Prolog: assertz(safe_mode_active(Reason)).
        Returns fact ID.
        """
        return self._bridge.assert_fact(
            "safe_mode_active",
            [reason],
            FactScope.PERSISTENT,
        )

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=["REQ-007-FR8-006", "REQ-007-US-002-010"],
    )
    def exit_safe_mode(self) -> bool:
        """Retract all safe_mode_active facts."""
        retracted = False
        # Retract via tracked facts in bridge
        fact_ids_to_remove = []
        for fact_id, fact in self._bridge._facts.items():
            if fact.predicate == "safe_mode_active":
                fact_ids_to_remove.append(fact_id)
        for fact_id in fact_ids_to_remove:
            if self._bridge.retract_fact(fact_id):
                retracted = True
        # Also try direct retraction for untracked facts
        if not retracted:
            try:
                list(self._bridge._prolog.query("retract(safe_mode_active(_))"))
                retracted = True
            except Exception as exc:
                _logger.warning(
                    "Direct safe_mode_active retraction failed: %s", exc, exc_info=True
                )
        return retracted

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=["REQ-007-US-002-010"],
    )
    def is_safe_mode(self) -> bool:
        """Check if safe mode is currently active in Prolog."""
        try:
            return self._bridge.query_bool("safe_mode_active(_)")
        except Exception as exc:
            _logger.warning("safe_mode_active check failed: %s", exc, exc_info=True)
            return False

    # ------------------------------------------------------------------
    # 9. evaluate_context/2
    # ------------------------------------------------------------------
    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=["REQ-007-US-001-003"],
    )
    def evaluate_context(self, action: str, context: dict[str, Any]) -> dict[str, Any]:
        """Evaluate action in context, returning full decision details.

        Maps to Prolog: evaluate_policy(Action, Resource, Decision).
        """
        action_atom = self._format_atom(action)
        resource = context.get("resource", context.get("path", "unknown"))
        resource_atom = self._format_atom(resource)
        result = self._bridge.query(
            f"evaluate_policy({action_atom}, {resource_atom}, Decision)"
        )
        if result.success and result.solutions:
            decision = result.solutions[0].get("Decision", "deny")
            if isinstance(decision, bytes):
                decision = decision.decode()
            return {
                "decision": str(decision),
                "action": action,
                "resource": resource,
                "execution_time_ms": result.execution_time_ms,
            }
        return {
            "decision": "deny",
            "action": action,
            "resource": resource,
            "reason": "no matching policy",
        }

    # ------------------------------------------------------------------
    # 10. collect_metrics/0
    # ------------------------------------------------------------------
    @traces(
        spec="007",
        story="US-PGOV-006",
        reqs=["REQ-007-US-006-003"],
    )
    def collect_metrics(self) -> dict[str, Any]:
        """Collect governance metrics from Prolog state.

        Returns counts of active facts, sessions, violations.
        """
        fact_counts = self._bridge.get_fact_count()
        violations = self._bridge.query("violation(_, _, _, _)")
        violation_count = len(violations.solutions) if violations.success else 0
        safe_mode = self.is_safe_mode()
        return {
            "fact_counts": fact_counts,
            "violation_count": violation_count,
            "safe_mode_active": safe_mode,
        }

    # ------------------------------------------------------------------
    # Utility: generate cache key from action + context
    # ------------------------------------------------------------------
    @traces(
        spec="007",
        story="FR-PGOV-001",
        reqs=["REQ-007-FR1-008", "REQ-007-FR9-007"],
    )
    def generate_cache_key(self, action: str, context: dict[str, Any]) -> str:
        """Generate a deterministic cache key for a governance query."""
        key_data = f"{action}:{sorted(context.items())}"
        return hashlib.sha256(key_data.encode()).hexdigest()[:32]