orchestrata_ultimate.infra.prolog_bridge

Reference › orchestrata_ultimate › infra › prolog_bridge

The Python interface to SWI-Prolog for governance (Spec 007a). PrologBridge wraps the janus engine from core.prolog_adapter to execute governance rules, manage facts and keep session and request state. It is imported by the gateway server, the policy layers in governance and the agents' own Prolog bridge, and it satisfies GovernancePrologBridgeProtocol from core.protocols.decoupling.

The constructor takes an optional rules directory. load_rules consults one .pl file and load_all_rules loads every file in the directory, returning a count. query returns a QueryResult (success flag, solutions, execution time, error text); query_one returns the first solution and query_bool a success flag; queries are given the governance module prefix by _add_governance_prefix. assert_fact records a PrologFact with a FactScope (core.enums) as session, request or persistent, an optional session id and an optional TTL, and returns a deterministic fact id from _generate_fact_id; retract_fact removes it. start_session/end_session and start_request/end_request manage lifecycles and remove the facts of their scope, cleanup_expired_facts drops facts past their TTL, and get_fact_count reports counts by scope. Strings and keys are cleaned by _sanitize_string and _sanitize_key, and _python_to_prolog converts Python values to Prolog terms. The bridge also imports a Prolog worker from the orchestrator.

At a glance

Class diagram

Dependency diagram

API

Classes

PrologFact (dataclass)

Decorators: dataclass · Line 39

Represents a Prolog fact with metadata.

QueryResult (dataclass)

Decorators: dataclass · Line 59

Result of a Prolog query.

PrologBridge

Line 73

Bridge to SWI-Prolog for governance rule execution.

Functions

Source

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

"""Prolog Bridge for Orchestrata Governance.

Provides a Python interface to SWI-Prolog for executing governance rules,
managing facts, and maintaining session/request state.

Key Features:
- Wraps the janus-swi Prolog engine (via core.prolog_adapter) for Prolog interaction
- Manages fact lifecycle with scoping (session, request, persistent)
- Supports TTL-based fact expiration
- Type conversion between Python and Prolog
- Session and request lifecycle management
- Deterministic fact ID generation
"""

# Spec: 007a - Prolog Governance

import hashlib
import logging
import re
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.enums import FactScope

# Module-level logger
_logger = logging.getLogger(__name__)


def _utc_now() -> datetime:
    """Get current UTC time with timezone info."""
    return datetime.now(UTC)


@dataclass
class PrologFact:
    """Represents a Prolog fact with metadata."""

    fact_id: str
    predicate: str
    args: list[Any]
    scope: FactScope
    session_id: str | None = None
    ttl_seconds: int | None = None
    created_at: datetime = field(default_factory=_utc_now)

    def is_expired(self) -> bool:
        """Check if fact has exceeded TTL."""
        if self.ttl_seconds is None:
            return False
        elapsed = (_utc_now() - self.created_at).total_seconds()
        return elapsed > self.ttl_seconds


@dataclass
class QueryResult:
    """Result of a Prolog query."""

    success: bool
    solutions: list[dict[str, Any]] = field(default_factory=list)
    execution_time_ms: float = 0.0
    error: str | None = None

    @property
    def results(self) -> list[dict[str, Any]]:
        """Backward-compatible alias for solutions."""
        return self.solutions


class PrologBridge:
    """Bridge to SWI-Prolog for governance rule execution.

    Manages Prolog facts with lifecycle scoping, TTL expiration,
    and provides high-level query methods.
    """

    def __init__(self, rules_dir: Path | None = None) -> None:
        """Initialize PrologBridge.

        Args:
            rules_dir: Directory containing .pl rule files
        """
        # Lazy import: prolog_worker lives in orchestrator/ (L4) — a
        # top-level import would be an L1->L4 upward edge.
        from orchestrata_ultimate.orchestrator.prolog_worker import get_prolog_worker

        self._worker = get_prolog_worker()
        self._rules_dir = rules_dir
        self._loaded_files: set[str] = set()
        self._facts: dict[str, PrologFact] = {}
        self._session_facts: dict[str, set[str]] = {}  # session_id -> fact_ids
        self._request_facts: set[str] = set()  # fact_ids for current request

    @property
    def _prolog(self) -> Any:
        """Backward-compatible access to the Prolog engine.

        Returns a PrologProxy that routes through the worker queue,
        or a test-injected override if one was set via the setter.
        """
        if hasattr(self, "_prolog_override") and self._prolog_override is not None:
            return self._prolog_override
        from orchestrata_ultimate.orchestrator.prolog_worker import PrologProxy

        return PrologProxy(self._worker)

    @_prolog.setter
    def _prolog(self, value: Any) -> None:
        """Allow tests to inject a replacement Prolog engine."""
        self._prolog_override = value

    # ===== Rule Loading =====

    def load_rules(self, rule_file: Path) -> bool:
        """Load a Prolog rule file.

        Args:
            rule_file: Path to .pl file

        Returns:
            True if loaded successfully, False otherwise
        """
        if not rule_file.exists():
            return False

        abs_path = str(rule_file.absolute())

        if abs_path in self._loaded_files:
            return True  # Already loaded

        try:
            self._prolog.consult(abs_path)
            self._loaded_files.add(abs_path)
            return True
        except Exception as e:
            _logger.error(f"Failed to load Prolog rules from {abs_path}: {e}")
            return False

    def load_all_rules(self) -> int:
        """Load all .pl files from rules directory.

        Returns:
            Number of files loaded
        """
        if self._rules_dir is None or not self._rules_dir.exists():
            return 0

        count = 0
        for rule_file in self._rules_dir.glob("*.pl"):
            if self.load_rules(rule_file):
                count += 1

        return count

    # Governance module predicates that need module prefix
    GOVERNANCE_PREDICATES = {
        "session_has_goals",
        "session_blacklisted",
        "session_active",
        "session_checklist_complete",
        "session_tests_passing",
        "session_artifacts_valid",
        "tool_blocked",
        "intent_action",
        "intent_resource",
        "intent_constraints",
        "volk_active",
        "volk_blacklisted",
        "volk_priority",
        "volk_has_budget",
        "volk_exists",
        "resource_available",
        "issue_severity",
        "issue_security_related",
        "action_type",
        "action_violates_policy",
        "action_in_protected_path",
        "termination_reason_valid",
        "volk_type_allowed",
        "volk_type_blacklisted",
        "resources_available",
        # Also include exported governance predicates for queries
        "can_open_session",
        "can_close_session",
        "session_valid",
        "session_requirements_met",
        "tool_allowed",
        "policy_permits",
        "tool_requires_approval",
        "escalation_target",
        "escalation_required",
        "escalation_path_valid",
        "can_allocate_resource",
        "priority_allocation",
        "resource_allocation_fair",
        "can_hire_volk",
        "can_fire_volk",
        "volk_operational",
        "audit_required",
        "compliance_check_passed",
        "compliance_violation",
        "evaluate_policy",
        "check_constraint",
        "protected_path",
        # Command classification predicates
        "safe_command",
        "destructive_command",
        "destructive_command_lower",
        "command_whitelisted",
        "command_blacklisted",
    }

    # Knowledge module predicates that need module prefix
    KNOWLEDGE_PREDICATES = {
        "owns",
        "depends_on",
        "agent_scope_node",
        "node_access_override",
    }

    def _add_governance_prefix(self, query_str: str) -> str:
        """Add governance module prefix to query if needed.

        Args:
            query_str: Original query string

        Returns:
            Query with governance: prefix if applicable
        """
        # Skip if already has a module prefix
        if ":" in query_str.split("(")[0]:
            return query_str

        # Check if query starts with a governance predicate
        for pred in self.GOVERNANCE_PREDICATES:
            if query_str.startswith(pred + "("):
                return f"governance:{query_str}"

        return query_str

    # ===== Query Execution =====

    def query(self, query_str: str) -> QueryResult:
        """Execute a Prolog query and return all solutions.

        Args:
            query_str: Prolog query string

        Returns:
            QueryResult with solutions and metadata
        """
        # Auto-add governance module prefix if needed
        query_str = self._add_governance_prefix(query_str)
        start_time = time.time()

        try:
            solutions = self._prolog.query(query_str)

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

            return QueryResult(
                success=len(solutions) > 0,
                solutions=solutions,
                execution_time_ms=execution_time,
            )
        except Exception as e:
            execution_time = (time.time() - start_time) * 1000
            return QueryResult(
                success=False,
                solutions=[],
                execution_time_ms=execution_time,
                error=str(e),
            )

    def query_one(self, query_str: str) -> dict[str, Any] | None:
        """Execute a query and return first solution.

        Args:
            query_str: Prolog query string

        Returns:
            First solution dict or None
        """
        result = self.query(query_str)
        if result.success and len(result.solutions) > 0:
            return result.solutions[0]
        return None

    def query_bool(self, query_str: str) -> bool:
        """Execute a query and return True if it succeeds.

        Args:
            query_str: Prolog query string

        Returns:
            True if query has at least one solution
        """
        result = self.query(query_str)
        return result.success

    # ===== Fact Management =====

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-011"])
    def assert_fact(
        self,
        predicate: str,
        args: list[Any],
        scope: FactScope,
        session_id: str | None = None,
        ttl_seconds: int | None = None,
    ) -> str:
        """Assert a Prolog fact with lifecycle management.

        Args:
            predicate: Predicate name
            args: List of arguments
            scope: Fact scope (session, request, persistent)
            session_id: Session ID for session-scoped facts
            ttl_seconds: Time-to-live in seconds

        Returns:
            Fact ID for later retraction
        """
        fact_id = self._generate_fact_id(predicate, args)

        # Convert args to Prolog format
        prolog_args = [self._python_to_prolog(arg) for arg in args]

        # Add module prefix for governance or knowledge predicates
        if predicate in self.GOVERNANCE_PREDICATES:
            fact_str = f"governance:{predicate}({', '.join(prolog_args)})"
        elif predicate in self.KNOWLEDGE_PREDICATES:
            fact_str = f"knowledge:{predicate}({', '.join(prolog_args)})"
        else:
            fact_str = f"{predicate}({', '.join(prolog_args)})"

        # Assert to Prolog via worker queue
        try:
            self._prolog.assertz(fact_str)
        except Exception as e:
            # Log the error but continue - fact might already exist
            _logger.debug(f"Prolog assertz failed for {fact_str}: {e}")

        # Track fact metadata
        fact = PrologFact(
            fact_id=fact_id,
            predicate=predicate,
            args=args,
            scope=scope,
            session_id=session_id,
            ttl_seconds=ttl_seconds,
        )

        self._facts[fact_id] = fact

        # Add to scope-specific tracking
        if scope == FactScope.SESSION and session_id:
            if session_id not in self._session_facts:
                self._session_facts[session_id] = set()
            self._session_facts[session_id].add(fact_id)
        elif scope == FactScope.REQUEST:
            self._request_facts.add(fact_id)

        return fact_id

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-011"])
    def retract_fact(self, fact_id: str) -> bool:
        """Retract a previously asserted fact.

        Args:
            fact_id: Fact ID from assert_fact

        Returns:
            True if retracted, False if not found
        """
        if fact_id not in self._facts:
            return False

        fact = self._facts[fact_id]

        # Convert args to Prolog format
        prolog_args = [self._python_to_prolog(arg) for arg in fact.args]

        # Add module prefix for governance or knowledge predicates
        if fact.predicate in self.GOVERNANCE_PREDICATES:
            fact_str = f"governance:{fact.predicate}({', '.join(prolog_args)})"
        elif fact.predicate in self.KNOWLEDGE_PREDICATES:
            fact_str = f"knowledge:{fact.predicate}({', '.join(prolog_args)})"
        else:
            fact_str = f"{fact.predicate}({', '.join(prolog_args)})"

        # Retract from Prolog via worker queue
        try:
            self._prolog.retract(fact_str)
        except Exception as e:
            # Log the error but continue - fact might already be retracted
            _logger.debug(f"Prolog retract failed for {fact_str}: {e}")

        # Remove from tracking
        del self._facts[fact_id]

        if fact.scope == FactScope.SESSION and fact.session_id:
            if fact.session_id in self._session_facts:
                self._session_facts[fact.session_id].discard(fact_id)
        elif fact.scope == FactScope.REQUEST:
            self._request_facts.discard(fact_id)

        return True

    def _generate_fact_id(self, predicate: str, args: list[Any]) -> str:
        """Generate deterministic fact ID.

        Args:
            predicate: Predicate name
            args: List of arguments

        Returns:
            Deterministic fact ID
        """
        content = f"{predicate}:{str(args)}"
        hash_obj = hashlib.sha256(content.encode())
        return f"fact-{hash_obj.hexdigest()[:16]}"

    # ===== Session Lifecycle =====

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-011"])
    def start_session(
        self, session_id: str, goals: list[str], context: dict[str, Any]
    ) -> None:
        """Start a new session with goals and context.

        Args:
            session_id: Unique session identifier
            goals: List of session goals
            context: Session context dictionary
        """
        # Assert session_active fact
        self.assert_fact(
            "session_active", [session_id], FactScope.SESSION, session_id=session_id
        )

        # Assert goals
        for goal in goals:
            self.assert_fact(
                "session_goal",
                [session_id, goal],
                FactScope.SESSION,
                session_id=session_id,
            )

        # Assert has_goals if goals present
        if goals:
            self.assert_fact(
                "session_has_goals",
                [session_id],
                FactScope.SESSION,
                session_id=session_id,
            )

        # Assert context facts
        for key, value in context.items():
            self.assert_fact(
                "session_context",
                [session_id, key, value],
                FactScope.SESSION,
                session_id=session_id,
            )

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-011"])
    def end_session(self, session_id: str) -> int:
        """End a session and clean up all session-scoped facts.

        Args:
            session_id: Session to end

        Returns:
            Number of facts retracted
        """
        if session_id not in self._session_facts:
            return 0

        fact_ids = list(self._session_facts[session_id])
        count = 0

        for fact_id in fact_ids:
            if self.retract_fact(fact_id):
                count += 1

        del self._session_facts[session_id]
        return count

    # ===== Request Lifecycle =====

    def start_request(self, session_id: str, tool: str, intent: str) -> None:
        """Start a new request with tool and intent.

        Args:
            session_id: Current session ID
            tool: Tool being executed
            intent: Intent classification
        """
        self.assert_fact(
            "current_request", [session_id, tool, intent], FactScope.REQUEST
        )

    def end_request(self) -> int:
        """End current request and clean up request-scoped facts.

        Returns:
            Number of facts retracted
        """
        fact_ids = list(self._request_facts)
        count = 0

        for fact_id in fact_ids:
            if self.retract_fact(fact_id):
                count += 1

        return count

    # ===== Fact Cleanup =====

    def cleanup_expired_facts(self) -> int:
        """Remove facts that have exceeded their TTL.

        Returns:
            Number of facts cleaned up
        """
        expired = []

        for fact_id, fact in self._facts.items():
            if fact.is_expired():
                expired.append(fact_id)

        count = 0
        for fact_id in expired:
            if self.retract_fact(fact_id):
                count += 1

        return count

    def get_fact_count(self) -> dict[str, int]:
        """Get count of facts by scope.

        Returns:
            Dict with counts for each scope
        """
        counts = {"session": 0, "request": 0, "persistent": 0}

        for fact in self._facts.values():
            counts[fact.scope.value] += 1

        return counts

    # ===== Type Conversion =====

    def _sanitize_string(self, value: str) -> str:
        """Sanitize a string for safe use in Prolog.

        Escapes all potentially dangerous characters to prevent injection.

        Args:
            value: Raw string value

        Returns:
            Sanitized string safe for Prolog
        """
        # Escape backslashes first (must be done before other escapes)
        result = value.replace("\\", "\\\\")
        # Escape single quotes (Prolog string delimiter)
        result = result.replace("'", "''")
        # Escape double quotes
        result = result.replace('"', '\\"')
        # Escape newlines and carriage returns
        result = result.replace("\n", "\\n")
        result = result.replace("\r", "\\r")
        # Escape tabs
        result = result.replace("\t", "\\t")
        # Escape backticks (can be special in some Prolog implementations)
        result = result.replace("`", "\\`")
        # Remove null bytes (security risk)
        result = result.replace("\x00", "")
        # Limit length to prevent DoS
        return result[:4096]

    def _sanitize_key(self, key: str) -> str:
        """Sanitize a dictionary key for use in Prolog.

        Args:
            key: Dictionary key

        Returns:
            Sanitized key safe for Prolog atoms
        """
        # Only allow alphanumeric and underscores in keys
        sanitized = re.sub(r"[^\w]", "_", str(key))
        # Ensure key starts with lowercase letter (Prolog atom requirement)
        if sanitized and not sanitized[0].islower():
            sanitized = "k_" + sanitized
        return sanitized[:256]

    def _python_to_prolog(self, value: Any) -> str:
        """Convert Python value to Prolog term string.

        Args:
            value: Python value

        Returns:
            Prolog term string with proper escaping
        """
        if isinstance(value, bool):
            return "true" if value else "false"
        elif isinstance(value, (int, float)):
            # Validate numeric values
            if not (-1e15 < value < 1e15):
                _logger.warning(f"Numeric value {value} out of safe range, clamping")
                value = max(-1e15, min(1e15, value))
            return str(value)
        elif isinstance(value, str):
            # Full sanitization for strings
            sanitized = self._sanitize_string(value)
            return f"'{sanitized}'"
        elif isinstance(value, list):
            # Limit list size to prevent DoS
            if len(value) > 1000:
                _logger.warning(
                    f"List too long ({len(value)} items), truncating to 1000"
                )
                value = value[:1000]
            prolog_items = [self._python_to_prolog(item) for item in value]
            return f"[{','.join(prolog_items)}]"
        elif isinstance(value, dict):
            # Limit dict size to prevent DoS
            if len(value) > 100:
                _logger.warning(
                    f"Dict too large ({len(value)} items), truncating to 100"
                )
                value = dict(list(value.items())[:100])
            # Convert dict to list of key-value pairs with sanitized keys
            pairs = [
                f"'{self._sanitize_key(k)}'-{self._python_to_prolog(v)}"
                for k, v in value.items()
            ]
            return f"[{','.join(pairs)}]"
        else:
            # Convert to string and sanitize
            sanitized = self._sanitize_string(str(value))
            return f"'{sanitized}'"