orchestrata_ultimate.session.prolog_facts

Reference › orchestrata_ultimate › session › prolog_facts

PrologFactBridge carries typed session Fact objects into and out of the Prolog knowledge base through a PrologBridge. The bridge is optional: set_bridge supplies it, bridge returns it and has_bridge says whether one is present. assert_fact asserts one fact as a Prolog term and returns the term (or nothing), batch_assert asserts several and returns the count, retract_fact removes one by id, clear_session_facts retracts all facts of a session and returns the count, query_facts reads a session's facts back, optionally by fact type, and asserted_count reports how many are currently asserted. It works at the level of facts; the session-level terms are the job of prolog_interface.

At a glance

Class diagram

Dependency diagram

API

Classes

PrologFactBridge

Decorators: traces(spec='006', story='US-001', reqs=['REQ-006-FR2-001']) · Line 27

Bridges session facts to/from Prolog knowledge base.

Source

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

# mypy: ignore-errors
"""Prolog fact bridge for session facts.

Converts Python Fact objects to Prolog terms and manages
assertion/retraction via PrologBridge.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.types import Fact

if TYPE_CHECKING:
    from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

logger = logging.getLogger(__name__)


@traces(
    spec="006",
    story="US-001",
    reqs=["REQ-006-FR2-001"],
)
class PrologFactBridge:
    """Bridges session facts to/from Prolog knowledge base."""

    def __init__(self, prolog_bridge: PrologBridge | None = None) -> None:
        self._bridge = prolog_bridge
        self._asserted_ids: set[str] = set()

    @property
    def bridge(self) -> PrologBridge | None:
        """Get the underlying PrologBridge."""
        return self._bridge

    def set_bridge(self, bridge: PrologBridge) -> None:
        """Set the PrologBridge instance."""
        self._bridge = bridge

    @property
    def has_bridge(self) -> bool:
        """Check if a PrologBridge is available."""
        return self._bridge is not None

    @traces(
        spec="006",
        story="US-001",
        reqs=["REQ-006-FR6-003", "REQ-006-US-001-004", "REQ-006-US-004-002"],
    )
    def assert_fact(self, session_id: str, fact: Fact) -> str | None:
        """Assert a session fact as a Prolog term.

        Returns:
            Fact ID if asserted, None if bridge unavailable.
        """
        if self._bridge is None:
            return None

        try:
            from orchestrata_ultimate.core.enums import FactScope

            fact_id = self._bridge.assert_fact(
                "session_fact",
                [session_id, fact.fact_type.value, fact.id],
                scope=FactScope.SESSION,
            )
            self._asserted_ids.add(fact.id)
            return fact_id
        except Exception as e:
            logger.warning("Failed to assert fact %s: %s", fact.id, e)
            return None

    @traces(spec="006", story="US-002", reqs=["REQ-006-US-002-003"])
    def retract_fact(self, session_id: str, fact_id: str) -> bool:
        """Retract a session fact from Prolog.

        Returns:
            True if retracted.
        """
        if self._bridge is None:
            return False

        try:
            self._check_safe_atom(session_id, "session_id")
            self._check_safe_atom(fact_id, "fact_id")
            # Find and retract by querying for the specific fact
            query = f"session_fact('{session_id}', _, '{fact_id}')"
            results = list(self._bridge._prolog.query(f"retract({query})"))
            self._asserted_ids.discard(fact_id)
            return len(results) > 0
        except Exception as e:
            logger.warning("Failed to retract fact %s: %s", fact_id, e)
            return False

    @traces(spec="006", story="FR-004", reqs=["REQ-006-FR4-004"])
    def query_facts(self, session_id: str, fact_type: str | None = None) -> list[dict]:
        """Query Prolog for session facts.

        Returns:
            List of matching fact dicts.
        """
        if self._bridge is None:
            return []

        try:
            self._check_safe_atom(session_id, "session_id")
            if fact_type:
                self._check_safe_atom(fact_type, "fact_type")
                query = f"session_fact('{session_id}', '{fact_type}', FactId)"
            else:
                query = f"session_fact('{session_id}', FactType, FactId)"

            results = self._bridge.query(query)
            if results.success:
                return results.results
            return []
        except Exception as e:
            logger.warning("Failed to query facts: %s", e)
            return []

    @traces(spec="006", story="US-004", reqs=["REQ-006-US-004-005"])
    def batch_assert(self, session_id: str, facts: list[Fact]) -> int:
        """Assert multiple facts in batch.

        Returns:
            Number of facts asserted.
        """
        if self._bridge is None:
            return 0

        count = 0
        for fact in facts:
            result = self.assert_fact(session_id, fact)
            if result is not None:
                count += 1
        return count

    @traces(spec="006", story="US-002", reqs=["REQ-006-FR5-001", "REQ-006-US-002-003"])
    def clear_session_facts(self, session_id: str) -> int:
        """Retract all facts for a session.

        Returns:
            Number of facts retracted.
        """
        if self._bridge is None:
            return 0

        try:
            self._check_safe_atom(session_id, "session_id")
            query = f"retract(session_fact('{session_id}', _, _))"
            count = 0
            while True:
                results = list(self._bridge._prolog.query(query))
                if not results:
                    break
                count += 1
            self._asserted_ids.clear()
            return count
        except Exception as e:
            logger.warning("Failed to clear session facts: %s", e)
            return 0

    @property
    def asserted_count(self) -> int:
        """Number of facts currently asserted."""
        return len(self._asserted_ids)

    @staticmethod
    def _check_safe_atom(value: str, field_name: str) -> None:
        """Reject values that would break out of a single-quoted Prolog atom.

        `retract_fact`, `query_facts`, and `clear_session_facts` interpolate
        `session_id`/`fact_id`/`fact_type` directly into raw Prolog query
        strings (unlike `assert_fact`, which routes its args through
        `PrologBridge`'s own `_python_to_prolog` escaping). A caller-supplied
        value containing a single quote would prematurely close the quoted
        atom and let arbitrary Prolog goals be appended to the query — a
        Prolog-injection risk. Reject such values instead of executing them.

        Raises:
            ValueError: If `value` contains a single quote.
        """
        if "'" in value:
            raise ValueError(f"Unsafe character in {field_name}: {value!r}")

    @traces(spec="006", story="FR-004", reqs=["REQ-006-FR4-001", "REQ-006-FR8-002"])
    def _fact_to_term(self, session_id: str, fact: Fact) -> str:
        """Convert a Fact to a Prolog term string."""
        fact_type = fact.fact_type.value
        return f"session_fact('{session_id}', '{fact_type}', '{fact.id}')"