orchestrata_ultimate.kernel.policy_evaluator

Reference › orchestrata_ultimate › kernel › policy_evaluator

The kernel's Prolog-backed policy evaluator. PolicyEvaluator asks the governance rule base four kinds of question: may this sender route this message type to this recipient, may a volk perform an action on a resource, does a volk own a component, and where should an issue be escalated. It serves SPEC-001 story US-003 (REQ-001-US-003-001 to -010). dispatcher imports it and uses it in audit-only mode on every dispatch. The rules themselves live in Prolog and are loaded through governance prolog_client; the broader decision engine is policy_engine.

• evaluate_routing runs the Prolog goal policy_permits/4 and returns a RoutingResult whose decision is allow, deny, queue or redirect. A deny is followed by an explanation helper.
• check_permission runs tool_allowed/4 and returns a PermissionResult with level allow, deny, require_approval or escalate.
• validate_ownership runs the ownership form of check_constraint/4 and returns an OwnershipResult with status valid, invalid, delegated or expired.
• get_escalation_path runs escalation_target/4 and returns an EscalationPath; with no answer, an error or a timeout it falls back to the orchestrator alias U:Ox, and it assigns a timeout of 60, 180, 300 or 600 seconds for critical, high, medium or low severity.
• dry_run builds the same query for a named type without side effects and returns a DryRunResult.

Queries run in a worker thread under asyncio.wait_for with a timeout (5 seconds by default); a timeout raises PolicyTimeoutError, any other failure raises PolicyQueryError, both subclasses of PolicyEvaluationError. For routing, _execute_routing_query catches these and turns them into a decision according to the governance strict flag from _strict: deny with a fail-closed reason when strict, allow with a permissive-fallback reason otherwise. Each result carries a provenance of policy or error. Results are cached for 60 seconds in a local dictionary and, when a Redis state manager is supplied, in Redis; invalidate_cache clears entries. Every query is recorded as a PolicyAuditEntry, readable with get_audit_log and cleared with clear_audit_log, and routing decisions are also reported through the governance verdict helper.

At a glance

Class diagram

Dependency diagram

API

Classes

RoutingDecision (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-001']) · Line 53

Result of a routing policy evaluation.

PermissionLevel (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-002']) · Line 63

Permission levels for actions.

OwnershipStatus (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-003']) · Line 73

Ownership validation status.

RoutingResult (dataclass)

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

Result of routing policy evaluation (REQ-001-US-003-001).

PermissionResult (dataclass)

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

Result of permission check (REQ-001-US-003-002).

OwnershipResult (dataclass)

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

Result of ownership validation (REQ-001-US-003-003).

EscalationPath (dataclass)

Decorators: dataclass, traces(spec='001', story='US-005', reqs=['REQ-001-US-005-011']) · Line 170

Escalation path for conflict resolution (REQ-001-US-003-004).

DryRunResult (dataclass)

Decorators: dataclass, traces(spec='001', story='US-003', reqs=['REQ-001-US-003-010']) · Line 199

Result of dry-run query (REQ-001-US-003-010).

PolicyAuditEntry (dataclass)

Decorators: dataclass, traces(spec='001', story='US-003', reqs=['REQ-001-US-003-009']) · Line 222

Audit log entry for policy queries (REQ-001-US-003-009).

PolicyEvaluationError

Bases: Exception · Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-008']) · Line 251

Base exception for policy evaluation errors.

PolicyTimeoutError

Bases: PolicyEvaluationError · Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-007']) · Line 256

Query timed out (REQ-001-US-003-007).

PolicyQueryError

Bases: PolicyEvaluationError · Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-008']) · Line 266

Query execution failed (REQ-001-US-003-008).

PolicyEvaluator

Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-001', 'REQ-001-US-003-002', 'REQ-001-US-003-003', 'REQ-001-US-… · Line 285

Evaluates governance policies via Prolog.

Source

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

# mypy: ignore-errors
"""Policy Evaluator for Orchestration Kernel.

This module implements governance policy evaluation via Prolog
as specified in US-003 of the Orchestration Kernel spec.

Requirements Implemented:
- REQ-001-US-003-001: Query Prolog for routing decisions
- REQ-001-US-003-002: Query Prolog for permission checks
- REQ-001-US-003-003: Query Prolog for ownership validation
- REQ-001-US-003-004: Query Prolog for escalation rules
- REQ-001-US-003-005: Cache Prolog query results for performance
- REQ-001-US-003-006: Invalidate cache on policy updates
- REQ-001-US-003-007: Handle Prolog query timeouts
- REQ-001-US-003-008: Handle Prolog query failures
- REQ-001-US-003-009: Log all policy queries for audit
- REQ-001-US-003-010: Support policy dry-run mode (testing)

Wiring status (2026-07-06, decision D3/W4.6): PolicyEvaluator was fully
built and unit-tested but had zero production callers. It is now wired into
kernel.dispatcher.MessageDispatcher, staged behind the ``policy_audit_mode``
constructor flag -- AUDIT-ONLY: every dispatch() evaluates evaluate_routing()
and logs the verdict at INFO, but never blocks or alters the dispatch
outcome, and any evaluator error is caught and logged at WARNING so the
governance layer can never break dispatch while it observes real traffic.
The flag defaults to False (zero behavior change). Promotion to an
enforcing mode -- where a DENY verdict actually blocks or redirects -- is a
later, explicit flip and is deliberately NOT implemented by this wiring.
"""

import asyncio
import hashlib
import logging
import time
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance._strict import governance_strict
from orchestrata_ultimate.governance.verdict import (
    Provenance,
    explain_policy_permits_deny,
    make_verdict,
)

logger = logging.getLogger(__name__)


@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
class RoutingDecision(Enum):
    """Result of a routing policy evaluation."""

    ALLOW = "allow"
    DENY = "deny"
    QUEUE = "queue"  # Queue for later delivery
    REDIRECT = "redirect"  # Redirect to different recipient


@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-002"])
class PermissionLevel(Enum):
    """Permission levels for actions."""

    ALLOW = "allow"
    DENY = "deny"
    REQUIRE_APPROVAL = "require_approval"
    ESCALATE = "escalate"


@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-003"])
class OwnershipStatus(Enum):
    """Ownership validation status."""

    VALID = "valid"
    INVALID = "invalid"
    DELEGATED = "delegated"
    EXPIRED = "expired"


@dataclass
@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
class RoutingResult:
    """Result of routing policy evaluation (REQ-001-US-003-001)."""

    decision: RoutingDecision
    reason: str
    redirect_to: str | None = None
    queue_priority: int | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    execution_time_ms: float = 0.0
    cached: bool = False
    # W1-P6: "policy" (a real rule fired) or "error" (a query error/timeout
    # made the fallback decide) -- see governance/verdict.py. Defaults to
    # "policy" so pre-P6 callers constructing this dataclass positionally
    # see no behavior change.
    provenance: str = "policy"

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "decision": self.decision.value,
            "reason": self.reason,
            "redirect_to": self.redirect_to,
            "queue_priority": self.queue_priority,
            "metadata": self.metadata,
            "execution_time_ms": self.execution_time_ms,
            "cached": self.cached,
            "provenance": self.provenance,
        }


@dataclass
@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-002"])
class PermissionResult:
    """Result of permission check (REQ-001-US-003-002)."""

    level: PermissionLevel
    reason: str
    required_approvers: list[str] = field(default_factory=list)
    constraints: list[str] = field(default_factory=list)
    execution_time_ms: float = 0.0
    cached: bool = False
    provenance: str = "policy"  # W1-P6: "policy" | "error" -- see verdict.py

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-002"])
    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "level": self.level.value,
            "reason": self.reason,
            "required_approvers": self.required_approvers,
            "constraints": self.constraints,
            "execution_time_ms": self.execution_time_ms,
            "cached": self.cached,
            "provenance": self.provenance,
        }


@dataclass
@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-003"])
class OwnershipResult:
    """Result of ownership validation (REQ-001-US-003-003)."""

    status: OwnershipStatus
    owner: str | None = None
    delegated_to: str | None = None
    reason: str = ""
    execution_time_ms: float = 0.0
    cached: bool = False
    provenance: str = "policy"  # W1-P6: "policy" | "error" -- see verdict.py

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-003"])
    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "status": self.status.value,
            "owner": self.owner,
            "delegated_to": self.delegated_to,
            "reason": self.reason,
            "execution_time_ms": self.execution_time_ms,
            "cached": self.cached,
            "provenance": self.provenance,
        }


@dataclass
@traces(spec="001", story="US-005", reqs=["REQ-001-US-005-011"])
class EscalationPath:
    """Escalation path for conflict resolution (REQ-001-US-003-004)."""

    target: str  # Who to escalate to
    priority: int  # Escalation priority
    path: list[str]  # Full escalation chain
    reason: str = ""
    timeout_seconds: int = 300  # Escalation timeout
    execution_time_ms: float = 0.0
    cached: bool = False
    provenance: str = "policy"  # W1-P6: "policy" | "error" -- see verdict.py

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-004"])
    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "target": self.target,
            "priority": self.priority,
            "path": self.path,
            "reason": self.reason,
            "timeout_seconds": self.timeout_seconds,
            "execution_time_ms": self.execution_time_ms,
            "cached": self.cached,
            "provenance": self.provenance,
        }


@dataclass
@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-010"])
class DryRunResult:
    """Result of dry-run query (REQ-001-US-003-010)."""

    query_type: str
    would_succeed: bool
    result_preview: dict[str, Any]
    prolog_query: str
    execution_time_ms: float = 0.0

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-010"])
    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "query_type": self.query_type,
            "would_succeed": self.would_succeed,
            "result_preview": self.result_preview,
            "prolog_query": self.prolog_query,
            "execution_time_ms": self.execution_time_ms,
        }


@dataclass
@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-009"])
class PolicyAuditEntry:
    """Audit log entry for policy queries (REQ-001-US-003-009)."""

    timestamp: datetime
    query_type: str
    query_params: dict[str, Any]
    result: dict[str, Any]
    execution_time_ms: float
    cached: bool
    session_id: str | None = None
    volk_id: str | None = None
    dry_run: bool = False

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "timestamp": self.timestamp.isoformat(),
            "query_type": self.query_type,
            "query_params": self.query_params,
            "result": self.result,
            "execution_time_ms": self.execution_time_ms,
            "cached": self.cached,
            "session_id": self.session_id,
            "volk_id": self.volk_id,
            "dry_run": self.dry_run,
        }


@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-008"])
class PolicyEvaluationError(Exception):
    """Base exception for policy evaluation errors."""


@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-007"])
class PolicyTimeoutError(PolicyEvaluationError):
    """Query timed out (REQ-001-US-003-007)."""

    def __init__(self, query: str, timeout_seconds: float) -> None:
        self.query = query
        self.timeout_seconds = timeout_seconds
        super().__init__(f"Policy query timed out after {timeout_seconds}s: {query}")


@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-008"])
class PolicyQueryError(PolicyEvaluationError):
    """Query execution failed (REQ-001-US-003-008)."""

    def __init__(self, query: str, error: str) -> None:
        self.query = query
        self.error = error
        super().__init__(f"Policy query failed: {error}")


@traces(
    spec="001",
    story="US-003",
    reqs=[
        "REQ-001-US-003-001",
        "REQ-001-US-003-002",
        "REQ-001-US-003-003",
        "REQ-001-US-003-004",
    ],
)
class PolicyEvaluator:
    """Evaluates governance policies via Prolog.

    This class implements policy evaluation for routing, permissions,
    ownership, and escalation using Prolog rules.

    Features:
    - Caching for performance (REQ-001-US-003-005)
    - Cache invalidation (REQ-001-US-003-006)
    - Timeout handling (REQ-001-US-003-007)
    - Error handling (REQ-001-US-003-008)
    - Audit logging (REQ-001-US-003-009)
    - Dry-run mode (REQ-001-US-003-010)
    """

    def __init__(
        self,
        prolog_bridge: Any,  # PrologBridge
        redis_state: Any | None = None,  # RedisStateManager
        cache_ttl_seconds: int = 60,
        query_timeout_seconds: float = 5.0,
        env: Mapping[str, str] | None = None,
    ) -> None:
        """Initialize PolicyEvaluator.

        Args:
            prolog_bridge: PrologBridge instance for Prolog queries.
            redis_state: Optional RedisStateManager for caching.
            cache_ttl_seconds: TTL for cached results (REQ-001-US-003-005).
            query_timeout_seconds: Timeout for Prolog queries (REQ-001-US-003-007).
            env: Optional explicit environment mapping for
                ``governance_strict()`` (F2 fail-open fix) -- read at call
                time, never mutated; defaults to ``os.environ`` when None,
                matching every other governance strictness seam
                (``governance/_strict.py``).
        """
        self._prolog = prolog_bridge
        self._redis_state = redis_state
        self._cache_ttl = cache_ttl_seconds
        self._query_timeout = query_timeout_seconds
        self._env = env
        self._audit_log: list[PolicyAuditEntry] = []
        self._local_cache: dict[str, tuple[Any, float]] = {}  # key -> (value, expiry)

        logger.info(
            f"PolicyEvaluator initialized: cache_ttl={cache_ttl_seconds}s, "
            f"query_timeout={query_timeout_seconds}s"
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-005"])
    def _generate_cache_key(self, query_type: str, **params) -> str:
        """Generate cache key for a query.

        Args:
            query_type: Type of query (routing, permission, etc.)
            **params: Query parameters.

        Returns:
            Cache key string.
        """
        content = f"{query_type}:{sorted(params.items())}"
        hash_val = hashlib.sha256(content.encode()).hexdigest()[:16]
        return f"policy:{query_type}:{hash_val}"

    @traces(
        spec="001", story="US-003", reqs=["REQ-001-US-003-005", "REQ-001-US-010-011"]
    )
    def _get_cached(self, cache_key: str) -> dict[str, Any] | None:
        """Get cached result (REQ-001-US-003-005).

        Args:
            cache_key: Cache key.

        Returns:
            Cached result or None.
        """
        # Try local cache first
        if cache_key in self._local_cache:
            value, expiry = self._local_cache[cache_key]
            if time.time() < expiry:
                logger.debug(f"Local cache hit: {cache_key}")
                return value
            else:
                del self._local_cache[cache_key]

        # Try Redis cache
        if self._redis_state is not None:
            try:
                cached = self._redis_state.cache_get(cache_key)
                if cached is not None:
                    logger.debug(f"Redis cache hit: {cache_key}")
                    # Store in local cache for faster subsequent access
                    self._local_cache[cache_key] = (
                        cached,
                        time.time() + self._cache_ttl,
                    )
                    return cached
            except Exception as e:
                logger.warning(f"Cache lookup failed: {e}")

        return None

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-005"])
    def _set_cached(self, cache_key: str, value: dict[str, Any]) -> None:
        """Store result in cache (REQ-001-US-003-005).

        Args:
            cache_key: Cache key.
            value: Value to cache.
        """
        # Store in local cache
        self._local_cache[cache_key] = (value, time.time() + self._cache_ttl)

        # Store in Redis
        if self._redis_state is not None:
            try:
                self._redis_state.cache_set(
                    cache_key, value, ttl_seconds=self._cache_ttl
                )
                logger.debug(f"Cached result: {cache_key}")
            except Exception as e:
                logger.warning(f"Cache store failed: {e}")

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-006"])
    def invalidate_cache(self, pattern: str | None = None) -> int:
        """Invalidate cached policy results (REQ-001-US-003-006).

        Args:
            pattern: Optional pattern to match (e.g., "routing" for routing queries).
                    If None, invalidates all policy cache.

        Returns:
            Number of entries invalidated.
        """
        count = 0

        # Invalidate local cache
        if pattern is None:
            count = len(self._local_cache)
            self._local_cache.clear()
        else:
            keys_to_delete = [k for k in self._local_cache if pattern in k]
            for key in keys_to_delete:
                del self._local_cache[key]
                count += 1

        logger.info(f"Invalidated {count} cached policy entries (pattern={pattern})")
        return count

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-009"])
    def _log_audit(
        self,
        query_type: str,
        params: dict[str, Any],
        result: dict[str, Any],
        execution_time_ms: float,
        cached: bool,
        session_id: str | None = None,
        volk_id: str | None = None,
        dry_run: bool = False,
    ) -> None:
        """Log policy query for audit (REQ-001-US-003-009).

        Args:
            query_type: Type of query.
            params: Query parameters.
            result: Query result.
            execution_time_ms: Execution time.
            cached: Whether result was from cache.
            session_id: Optional session ID.
            volk_id: Optional Volk ID.
            dry_run: Whether this was a dry-run.
        """
        entry = PolicyAuditEntry(
            timestamp=datetime.now(UTC),
            query_type=query_type,
            query_params=params,
            result=result,
            execution_time_ms=execution_time_ms,
            cached=cached,
            session_id=session_id,
            volk_id=volk_id,
            dry_run=dry_run,
        )

        self._audit_log.append(entry)

        # Keep audit log bounded
        if len(self._audit_log) > 10000:
            self._audit_log = self._audit_log[-5000:]

        # Log for external audit systems
        logger.info(
            f"AUDIT: {query_type} "
            f"params={params} "
            f"result={result.get('decision', result.get('level', 'N/A'))} "
            f"time={execution_time_ms:.2f}ms "
            f"cached={cached} "
            f"dry_run={dry_run}"
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-007"])
    async def _execute_with_timeout(
        self,
        query: str,
        timeout: float | None = None,
    ) -> dict[str, Any]:
        """Execute Prolog query with timeout (REQ-001-US-003-007).

        Args:
            query: Prolog query string.
            timeout: Optional timeout override.

        Returns:
            Query result.

        Raises:
            PolicyTimeoutError: If query times out.
            PolicyQueryError: If query fails.
        """
        timeout = timeout or self._query_timeout

        try:
            # Run query in thread pool to allow timeout
            result = await asyncio.wait_for(
                asyncio.to_thread(self._prolog.query, query),
                timeout=timeout,
            )
            return {
                "success": result.success,
                "solutions": result.solutions,
                "error": result.error,
            }
        except TimeoutError:
            raise PolicyTimeoutError(query, timeout)
        except Exception as e:
            raise PolicyQueryError(query, str(e))

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-008"])
    def _execute_sync(self, query: str) -> dict[str, Any]:
        """Execute Prolog query synchronously (REQ-001-US-003-008).

        Args:
            query: Prolog query string.

        Returns:
            Query result.

        Raises:
            PolicyQueryError: If query fails.
        """
        try:
            result = self._prolog.query(query)
            return {
                "success": result.success,
                "solutions": result.solutions,
                "error": result.error,
            }
        except Exception as e:
            raise PolicyQueryError(query, str(e))

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
    def _routing_decision_from_query(
        self, query_result: dict[str, Any]
    ) -> tuple[RoutingDecision, str]:
        """Parse routing decision and reason from Prolog query result.

        NOTE (W1-P6): signature/return-shape is UNCHANGED from pre-P6 --
        this private method is directly unit-tested with strict tuple
        equality (tests/kernel/test_policy_evaluator.py) that this brief may
        not edit. Provenance is derived by the CALLER (evaluate_routing)
        from the existing "error:"-prefix convention on ``reason`` instead
        of a new return element.

        Args:
            query_result: Result from _execute_with_timeout.

        Returns:
            Tuple of (RoutingDecision, reason string).
        """
        if query_result["success"] and query_result["solutions"]:
            solution = query_result["solutions"][0]
            decision_str = str(solution.get("Decision", "allow")).lower()

            if decision_str == "deny":
                return RoutingDecision.DENY, "Policy denied routing"
            elif decision_str == "queue":
                return RoutingDecision.QUEUE, "Recipient unavailable, message queued"
            elif decision_str == "redirect":
                return RoutingDecision.REDIRECT, "Message redirected by policy"
            else:
                return RoutingDecision.ALLOW, "Policy allows routing"
        else:
            # F2 fail-open fix: this branch used to hardcode ALLOW
            # regardless of *why* no decision came back -- indistinguishable
            # from a genuine policy allow in the audit log. Route through
            # governance_strict() like every other governance error branch
            # (deny under strict, the default) and prefix the reason with
            # "error:" so it is never confused with a real "policy:"-shaped
            # verdict (this prefix IS the P6 verdict-provenance signal --
            # see evaluate_routing below).
            detail = query_result.get("error") or "no policy result"
            if governance_strict(self._env):
                return RoutingDecision.DENY, f"error: {detail} — fail-closed"
            return RoutingDecision.ALLOW, f"error: {detail} — permissive fallback"

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
    async def _execute_routing_query(
        self, from_volk: str, to_volk: str, message_type: str
    ) -> tuple[RoutingDecision, str]:
        """Execute Prolog routing query with error handling.

        Args:
            from_volk: Sender Volk alias.
            to_volk: Recipient Volk alias.
            message_type: Type of message being routed.

        Returns:
            Tuple of (RoutingDecision, reason string).
        """
        query = (
            f"policy_permits('{from_volk}', '{to_volk}', '{message_type}', Decision)"
        )
        try:
            query_result = await self._execute_with_timeout(query)
            return self._routing_decision_from_query(query_result)
        except PolicyTimeoutError:
            # REQ-001-US-003-007: timeout is an ERROR outcome, never a policy
            # verdict — route through governance_strict() exactly like the
            # no-result branches (W1-P2) and the sibling permission path:
            # deny under strict (the default), allow only under the explicit
            # permissive posture. Closed as the W1-P6 ERROR_ALLOW finding.
            logger.warning(
                f"Routing policy query timed out for {from_volk} -> {to_volk}"
            )
            if governance_strict(self._env):
                return (
                    RoutingDecision.DENY,
                    "error: Policy query timeout — fail-closed",
                )
            return (
                RoutingDecision.ALLOW,
                "error: Policy query timeout — permissive fallback",
            )
        except PolicyQueryError as e:
            # REQ-001-US-003-008: same ERROR-outcome discipline as the timeout
            # branch — strict denies, permissive allows, both labelled.
            logger.error(f"Routing policy query failed: {e}")
            if governance_strict(self._env):
                return (
                    RoutingDecision.DENY,
                    f"error: Policy query error: {e.error} — fail-closed",
                )
            return (
                RoutingDecision.ALLOW,
                f"error: Policy query error: {e.error} — permissive fallback",
            )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
    async def evaluate_routing(
        self,
        from_volk: str,
        to_volk: str,
        message_type: str,
        session_id: str | None = None,
    ) -> RoutingResult:
        """Evaluate routing policy (REQ-001-US-003-001).

        Query Prolog for routing decision based on sender, recipient,
        and message type.

        Args:
            from_volk: Sender Volk alias.
            to_volk: Recipient Volk alias.
            message_type: Type of message being routed.
            session_id: Optional session ID for context.

        Returns:
            RoutingResult with decision and metadata.

        Raises:
            PolicyTimeoutError: If query times out.
            PolicyQueryError: If query fails.
        """
        start_time = time.time()
        params = {
            "from_volk": from_volk,
            "to_volk": to_volk,
            "message_type": message_type,
        }

        # REQ-001-US-003-005: Check cache
        cache_key = self._generate_cache_key("routing", **params)
        cached = self._get_cached(cache_key)

        if cached is not None:
            execution_time = (time.time() - start_time) * 1000
            result = RoutingResult(
                decision=RoutingDecision(cached["decision"]),
                reason=cached["reason"],
                redirect_to=cached.get("redirect_to"),
                queue_priority=cached.get("queue_priority"),
                metadata=cached.get("metadata", {}),
                execution_time_ms=execution_time,
                cached=True,
                provenance=cached.get("provenance", "policy"),
            )
            self._log_audit(
                "routing", params, result.to_dict(), execution_time, True, session_id
            )
            return result

        decision, reason = await self._execute_routing_query(
            from_volk, to_volk, message_type
        )
        # W1-P6: the "error:"-prefix convention (already established by the
        # F2 fail-open fix, above) is the ONLY caller-visible signal --
        # _routing_decision_from_query/_execute_routing_query keep their
        # pre-P6 2-tuple signature/return exactly (they are directly
        # unit-tested with strict tuple equality this brief may not edit).
        is_error = reason.startswith("error:")
        if decision is RoutingDecision.DENY and not is_error:
            explanation = explain_policy_permits_deny(
                self._prolog.query_bool, from_volk, to_volk, message_type
            )
            reason = f"{reason} — {explanation}"
        provenance = Provenance.ERROR if is_error else Provenance.POLICY

        execution_time = (time.time() - start_time) * 1000
        result = RoutingResult(
            decision=decision,
            reason=reason,
            execution_time_ms=execution_time,
            cached=False,
            provenance=provenance.value,
        )

        # REQ-001-US-003-005: Cache result
        self._set_cached(cache_key, result.to_dict())

        # REQ-001-US-003-009: Audit log
        self._log_audit(
            "routing", params, result.to_dict(), execution_time, False, session_id
        )

        # W1-P6: structured verdict-provenance log (surface = evaluate_routing).
        make_verdict(
            surface="PolicyEvaluator.evaluate_routing",
            decision=decision.value,
            is_deny=decision is RoutingDecision.DENY,
            is_error=provenance is Provenance.ERROR,
            rule_source=(
                "error:policy_permits/4"
                if provenance is Provenance.ERROR
                else "policy_permits/4"
            ),
            reason=reason,
            logger_=logger,
        )

        return result

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-002"])
    def _permission_level_from_query(
        self, query_result: dict[str, Any]
    ) -> tuple[PermissionLevel, str]:
        """Parse permission level and reason from Prolog query result.

        NOTE (W1-P6): signature/return-shape is UNCHANGED from pre-P6 --
        see _routing_decision_from_query's matching note. Provenance is
        derived by the caller (check_permission) from the existing
        "error:"-prefix convention on ``reason``.

        Args:
            query_result: Result from _execute_with_timeout.

        Returns:
            Tuple of (PermissionLevel, reason string).
        """
        if query_result["success"] and query_result["solutions"]:
            solution = query_result["solutions"][0]
            level_str = str(solution.get("Level", "allow")).lower()

            if level_str == "deny":
                return PermissionLevel.DENY, "Permission denied by policy"
            elif level_str == "require_approval":
                return PermissionLevel.REQUIRE_APPROVAL, "Approval required"
            elif level_str == "escalate":
                return PermissionLevel.ESCALATE, "Escalation required"
            else:
                return PermissionLevel.ALLOW, "Permission granted"
        else:
            # F2 fail-open fix: see _routing_decision_from_query's matching
            # comment -- this branch used to hardcode ALLOW, inconsistent
            # with this same method's OWN timeout/query-error paths just
            # below (both already default to DENY). governance_strict()
            # makes all three error paths agree.
            detail = query_result.get("error") or "no policy result"
            if governance_strict(self._env):
                return PermissionLevel.DENY, f"error: {detail} — fail-closed"
            return PermissionLevel.ALLOW, f"error: {detail} — permissive fallback"

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-002"])
    async def _execute_permission_query(
        self, volk_id: str, action: str, resource: str
    ) -> tuple[PermissionLevel, str]:
        """Execute Prolog permission query with error handling.

        Args:
            volk_id: Volk requesting permission.
            action: Action to perform.
            resource: Resource being accessed.

        Returns:
            Tuple of (PermissionLevel, reason string).
        """
        query = f"tool_allowed('{volk_id}', '{action}', '{resource}', Level)"
        try:
            query_result = await self._execute_with_timeout(query)
            return self._permission_level_from_query(query_result)
        except PolicyTimeoutError:
            logger.warning(
                f"Permission check timed out for {volk_id}/{action}/{resource}"
            )
            return (
                PermissionLevel.DENY,
                "error: Policy query timeout, defaulting to deny for safety",
            )
        except PolicyQueryError as e:
            logger.error(f"Permission check failed: {e}")
            return PermissionLevel.DENY, f"error: Policy query error: {e.error}"

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-002"])
    async def check_permission(
        self,
        volk_id: str,
        action: str,
        resource: str,
        session_id: str | None = None,
    ) -> PermissionResult:
        """Check permission for action (REQ-001-US-003-002).

        Query Prolog for permission to perform action on resource.

        Args:
            volk_id: Volk requesting permission.
            action: Action to perform (read, write, execute, etc.).
            resource: Resource being accessed.
            session_id: Optional session ID.

        Returns:
            PermissionResult with level and constraints.

        Raises:
            PolicyTimeoutError: If query times out.
            PolicyQueryError: If query fails.
        """
        start_time = time.time()
        params = {
            "volk_id": volk_id,
            "action": action,
            "resource": resource,
        }

        # REQ-001-US-003-005: Check cache
        cache_key = self._generate_cache_key("permission", **params)
        cached = self._get_cached(cache_key)

        if cached is not None:
            execution_time = (time.time() - start_time) * 1000
            result = PermissionResult(
                level=PermissionLevel(cached["level"]),
                reason=cached["reason"],
                required_approvers=cached.get("required_approvers", []),
                constraints=cached.get("constraints", []),
                execution_time_ms=execution_time,
                cached=True,
                provenance=cached.get("provenance", "policy"),
            )
            self._log_audit(
                "permission",
                params,
                result.to_dict(),
                execution_time,
                True,
                session_id,
                volk_id,
            )
            return result

        level, reason = await self._execute_permission_query(volk_id, action, resource)
        # W1-P6: same "error:"-prefix convention as evaluate_routing --
        # _permission_level_from_query/_execute_permission_query keep their
        # pre-P6 2-tuple signature/return (directly unit-tested elsewhere).
        is_error = reason.startswith("error:")
        provenance = Provenance.ERROR if is_error else Provenance.POLICY

        execution_time = (time.time() - start_time) * 1000
        result = PermissionResult(
            level=level,
            reason=reason,
            execution_time_ms=execution_time,
            cached=False,
            provenance=provenance.value,
        )

        # REQ-001-US-003-005: Cache result
        self._set_cached(cache_key, result.to_dict())

        # REQ-001-US-003-009: Audit log
        self._log_audit(
            "permission",
            params,
            result.to_dict(),
            execution_time,
            False,
            session_id,
            volk_id,
        )

        # W1-P6: structured verdict-provenance log.
        make_verdict(
            surface="PolicyEvaluator.check_permission",
            decision=level.value,
            is_deny=level is PermissionLevel.DENY,
            is_error=provenance is Provenance.ERROR,
            rule_source=(
                "error:tool_allowed/4"
                if provenance is Provenance.ERROR
                else "tool_allowed/4"
            ),
            reason=reason,
            logger_=logger,
        )

        return result

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-003"])
    async def validate_ownership(
        self,
        volk_id: str,
        component: str,
        session_id: str | None = None,
    ) -> OwnershipResult:
        """Validate ownership of component (REQ-001-US-003-003).

        Query Prolog to verify if Volk owns or has rights to component.

        Args:
            volk_id: Volk claiming ownership.
            component: Component being validated.
            session_id: Optional session ID.

        Returns:
            OwnershipResult with status and owner info.

        Raises:
            PolicyTimeoutError: If query times out.
            PolicyQueryError: If query fails.
        """
        start_time = time.time()
        params = {
            "volk_id": volk_id,
            "component": component,
        }

        # REQ-001-US-003-005: Check cache
        cache_key = self._generate_cache_key("ownership", **params)
        cached = self._get_cached(cache_key)

        if cached is not None:
            execution_time = (time.time() - start_time) * 1000
            result = OwnershipResult(
                status=OwnershipStatus(cached["status"]),
                owner=cached.get("owner"),
                delegated_to=cached.get("delegated_to"),
                reason=cached.get("reason", ""),
                execution_time_ms=execution_time,
                cached=True,
                provenance=cached.get("provenance", "policy"),
            )
            self._log_audit(
                "ownership",
                params,
                result.to_dict(),
                execution_time,
                True,
                session_id,
                volk_id,
            )
            return result

        # Build Prolog query
        query = f"check_constraint(ownership, '{volk_id}', '{component}', Owner)"
        provenance = Provenance.POLICY

        try:
            query_result = await self._execute_with_timeout(query)

            if query_result["success"] and query_result["solutions"]:
                solution = query_result["solutions"][0]
                owner = str(solution.get("Owner", ""))

                if owner == volk_id:
                    status = OwnershipStatus.VALID
                    reason = f"{volk_id} owns {component}"
                elif owner:
                    status = OwnershipStatus.INVALID
                    reason = f"{component} is owned by {owner}, not {volk_id}"
                else:
                    status = OwnershipStatus.VALID
                    reason = "No ownership restriction"
            else:
                # No ownership rule, default to valid. W1-P6: a genuine
                # query error (existence_error, ...) is distinguishable from
                # a real "no solutions" via query_result["error"].
                status = OwnershipStatus.VALID
                owner = None
                reason = "No ownership restriction found"
                if query_result.get("error"):
                    provenance = Provenance.ERROR

        except PolicyTimeoutError:
            logger.warning(f"Ownership validation timed out for {volk_id}/{component}")
            status = OwnershipStatus.INVALID
            owner = None
            reason = "Policy query timeout"
            provenance = Provenance.ERROR
        except PolicyQueryError as e:
            logger.error(f"Ownership validation failed: {e}")
            status = OwnershipStatus.INVALID
            owner = None
            reason = f"Policy query error: {e.error}"
            provenance = Provenance.ERROR

        execution_time = (time.time() - start_time) * 1000
        result = OwnershipResult(
            status=status,
            owner=owner,
            reason=reason,
            execution_time_ms=execution_time,
            cached=False,
            provenance=provenance.value,
        )

        # REQ-001-US-003-005: Cache result
        self._set_cached(cache_key, result.to_dict())

        # REQ-001-US-003-009: Audit log
        self._log_audit(
            "ownership",
            params,
            result.to_dict(),
            execution_time,
            False,
            session_id,
            volk_id,
        )

        # W1-P6: structured verdict-provenance log.
        make_verdict(
            surface="PolicyEvaluator.validate_ownership",
            decision=status.value,
            is_deny=status is OwnershipStatus.INVALID,
            is_error=provenance is Provenance.ERROR,
            rule_source=(
                "error:check_constraint/4"
                if provenance is Provenance.ERROR
                else "check_constraint/4"
            ),
            reason=reason,
            logger_=logger,
        )

        return result

    @traces(
        spec="001",
        story="US-003",
        reqs=["REQ-001-US-003-004", "REQ-001-US-003-022", "REQ-001-US-005-011"],
    )
    async def get_escalation_path(
        self,
        issue_type: str,
        severity: str,
        session_id: str | None = None,
        volk_id: str | None = None,
    ) -> EscalationPath:
        """Get escalation path for issue (REQ-001-US-003-004).

        Query Prolog for escalation rules based on issue type and severity.

        Args:
            issue_type: Type of issue (conflict, permission, resource, etc.).
            severity: Issue severity (low, medium, high, critical).
            session_id: Optional session ID.
            volk_id: Optional requesting Volk ID.

        Returns:
            EscalationPath with target and chain.

        Raises:
            PolicyTimeoutError: If query times out.
            PolicyQueryError: If query fails.
        """
        start_time = time.time()
        params = {
            "issue_type": issue_type,
            "severity": severity,
        }

        # REQ-001-US-003-005: Check cache
        cache_key = self._generate_cache_key("escalation", **params)
        cached = self._get_cached(cache_key)

        if cached is not None:
            execution_time = (time.time() - start_time) * 1000
            result = EscalationPath(
                target=cached["target"],
                priority=cached["priority"],
                path=cached["path"],
                reason=cached.get("reason", ""),
                timeout_seconds=cached.get("timeout_seconds", 300),
                execution_time_ms=execution_time,
                cached=True,
                provenance=cached.get("provenance", "policy"),
            )
            self._log_audit(
                "escalation",
                params,
                result.to_dict(),
                execution_time,
                True,
                session_id,
                volk_id,
            )
            return result

        # Build Prolog query
        query = f"escalation_target('{issue_type}', '{severity}', Target, Priority)"
        provenance = Provenance.POLICY

        try:
            query_result = await self._execute_with_timeout(query)

            if query_result["success"] and query_result["solutions"]:
                solution = query_result["solutions"][0]
                target = str(solution.get("Target", "U:Ox"))
                priority = int(solution.get("Priority", 1))
                path = [target]  # Default path is just target
                reason = f"Escalating {issue_type} ({severity}) to {target}"
            else:
                # Default escalation to Orchestrator. W1-P6: a genuine query
                # error is distinguishable from a real "no solutions" via
                # query_result["error"].
                target = "U:Ox"
                priority = 1
                path = ["U:Ox"]
                reason = "Default escalation to Orchestrator"
                if query_result.get("error"):
                    provenance = Provenance.ERROR

        except PolicyTimeoutError:
            logger.warning(f"Escalation query timed out for {issue_type}/{severity}")
            target = "U:Ox"
            priority = 1
            path = ["U:Ox"]
            reason = "Policy query timeout, default escalation"
            provenance = Provenance.ERROR
        except PolicyQueryError as e:
            logger.error(f"Escalation query failed: {e}")
            target = "U:Ox"
            priority = 1
            path = ["U:Ox"]
            reason = f"Policy query error: {e.error}"
            provenance = Provenance.ERROR

        execution_time = (time.time() - start_time) * 1000

        # Determine timeout based on severity
        timeout_map = {
            "critical": 60,
            "high": 180,
            "medium": 300,
            "low": 600,
        }
        timeout_seconds = timeout_map.get(severity.lower(), 300)

        result = EscalationPath(
            target=target,
            priority=priority,
            path=path,
            reason=reason,
            timeout_seconds=timeout_seconds,
            execution_time_ms=execution_time,
            cached=False,
            provenance=provenance.value,
        )

        # REQ-001-US-003-005: Cache result
        self._set_cached(cache_key, result.to_dict())

        # REQ-001-US-003-009: Audit log
        self._log_audit(
            "escalation",
            params,
            result.to_dict(),
            execution_time,
            False,
            session_id,
            volk_id,
        )

        # W1-P6: structured verdict-provenance log. Escalation always picks
        # A target (never a hard deny), so is_deny is always False here.
        make_verdict(
            surface="PolicyEvaluator.get_escalation_path",
            decision=f"escalate_to:{target}",
            is_deny=False,
            is_error=provenance is Provenance.ERROR,
            rule_source=(
                "error:escalation_target/4"
                if provenance is Provenance.ERROR
                else "escalation_target/4"
            ),
            reason=reason,
            logger_=logger,
        )

        return result

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-010"])
    async def dry_run(
        self,
        query_type: str,
        **params,
    ) -> DryRunResult:
        """Execute policy query in dry-run mode (REQ-001-US-003-010).

        Tests query without side effects or caching.

        Args:
            query_type: Type of query (routing, permission, ownership, escalation).
            **params: Query-specific parameters.

        Returns:
            DryRunResult with preview of what would happen.
        """
        start_time = time.time()

        # Build query based on type
        if query_type == "routing":
            query = (
                f"policy_permits('{params.get('from_volk', '')}', "
                f"'{params.get('to_volk', '')}', "
                f"'{params.get('message_type', '')}', Decision)"
            )
        elif query_type == "permission":
            query = (
                f"tool_allowed('{params.get('volk_id', '')}', "
                f"'{params.get('action', '')}', "
                f"'{params.get('resource', '')}', Level)"
            )
        elif query_type == "ownership":
            query = (
                f"check_constraint(ownership, '{params.get('volk_id', '')}', "
                f"'{params.get('component', '')}', Owner)"
            )
        elif query_type == "escalation":
            query = (
                f"escalation_target('{params.get('issue_type', '')}', "
                f"'{params.get('severity', '')}', Target, Priority)"
            )
        else:
            query = f"unknown_query_type({query_type})"

        try:
            query_result = self._execute_sync(query)
            would_succeed = query_result["success"]
            result_preview = {
                "solutions": query_result["solutions"],
                "error": query_result["error"],
            }
        except Exception as e:
            would_succeed = False
            result_preview = {"error": str(e)}

        execution_time = (time.time() - start_time) * 1000

        result = DryRunResult(
            query_type=query_type,
            would_succeed=would_succeed,
            result_preview=result_preview,
            prolog_query=query,
            execution_time_ms=execution_time,
        )

        # REQ-001-US-003-009: Log dry-run in audit
        self._log_audit(
            f"dry_run:{query_type}",
            params,
            result.to_dict(),
            execution_time,
            False,
            dry_run=True,
        )

        return result

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-009"])
    def get_audit_log(
        self,
        limit: int = 100,
        query_type: str | None = None,
        session_id: str | None = None,
    ) -> list[PolicyAuditEntry]:
        """Get audit log entries (REQ-001-US-003-009).

        Args:
            limit: Maximum entries to return.
            query_type: Optional filter by query type.
            session_id: Optional filter by session ID.

        Returns:
            List of audit entries.
        """
        entries = self._audit_log

        if query_type is not None:
            entries = [e for e in entries if e.query_type == query_type]

        if session_id is not None:
            entries = [e for e in entries if e.session_id == session_id]

        # Return most recent entries
        return entries[-limit:]

    def clear_audit_log(self) -> int:
        """Clear audit log.

        Returns:
            Number of entries cleared.
        """
        count = len(self._audit_log)
        self._audit_log.clear()
        logger.info(f"Cleared {count} audit log entries")
        return count