orchestrata_ultimate.session.prolog_interface

Reference › orchestrata_ultimate › session › prolog_interface

PrologSessionInterface is the bridge between sessions and the Prolog engine (spec 006), where Prolog is the source of truth for session state. The engine is attached with set_engine. assert_session_open asserts session(SessionID, active, '', Timestamp), retract_session removes the session term, and assert_session_fact asserts session_fact(SessionID, FactTerm, Timestamp, Source); the FactEmitter calls the latter in flush_to_prolog. Reads are query_session, query_session_facts, query_sessions and query_work by criteria, fact_exists and execute_query for an arbitrary query. The facts generated from context.json by FactGenerator are loaded with sync_facts and reload_facts, kept in a cache returned by get_cached_facts, written to a file by export_facts and removed with clear_facts; stats reports counters.

At a glance

Class diagram

Dependency diagram

API

Classes

PrologSessionInterface

Line 29

Interface between sessions and Prolog.

Source

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

# mypy: ignore-errors
"""Prolog session interface — spec 006 source-of-truth bridge.

Bridges session management with Prolog engine, maintaining four
core predicates as the authoritative session state:

    session(SessionID, Status, Model, Timestamp)
        Asserted when a session opens; retracted when it closes. Shares
        the canonical ``session/4`` arity with bulk-synced session facts
        so a single predicate covers assert/query/clear.

    session_fact(SessionID, FactTerm, Timestamp, Source)
        Asserted each time a fact is emitted during a session.

All timestamps are Unix epoch integers.
"""

import logging
import time
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.facts import FactGenerator

logger = logging.getLogger(__name__)


class PrologSessionInterface:
    """Interface between sessions and Prolog.

    Maintains session state as Prolog facts (source-of-truth) and
    provides query/sync helpers for the governance engine.

    Args:
        root: Project root directory used by ``FactGenerator``.
        engine: Optional Prolog engine instance (e.g. ``PrologEngine``
            or ``PrologBridge``). Can be set later via :meth:`set_engine`.

    Attributes:
        root: Project root as a ``Path``.
        engine: The Prolog engine, or ``None`` if not connected.
        fact_gen: ``FactGenerator`` used to produce session Prolog terms.
        facts_cache: Last batch of generated fact strings.

    Example:
        >>> iface = PrologSessionInterface("/tmp/project")
        >>> iface.assert_session_open("session_20260223_141500_abcd1234")
        True
    """

    def __init__(self, root: str | Path, engine: Any = None) -> None:
        """Initialize interface.

        Args:
            root: Project root directory.
            engine: Optional Prolog engine (PrologEngine or PrologBridge).
        """
        self.root = Path(root)
        self.engine = engine
        self.fact_gen = FactGenerator(root)
        self.facts_cache: list[str] = []

    @traces(
        spec="007", story="US-007", reqs=["REQ-007-US-003-008", "REQ-007-US-004-008"]
    )
    def set_engine(self, engine: Any) -> None:
        """Set Prolog engine.

        Args:
            engine: Prolog engine instance.
        """
        self.engine = engine

    # ------------------------------------------------------------------
    # Spec 006 Prolog source-of-truth predicates
    # ------------------------------------------------------------------

    @traces(
        spec="006",
        story="US-001",
        reqs=["REQ-006-FR2-001", "REQ-006-FR6-003", "REQ-006-US-001-004"],
    )
    def assert_session_open(self, session_id: str) -> bool:
        """Assert ``session(SessionID, active, '', Timestamp)`` in Prolog.

        Uses the canonical ``session/4`` predicate
        (``session(Seq, Purpose, Model, Timestamp)``) so that runtime
        open/retract/query and the bulk-synced facts share one predicate
        and :meth:`clear_facts` actually removes opened sessions.

        Called when a session is created. The fact remains in the
        knowledge base until :meth:`retract_session` is called.

        Args:
            session_id: The session identifier (spec 006 format).

        Returns:
            ``True`` if the fact was asserted, ``False`` if no engine.

        Example:
            >>> iface.assert_session_open("session_20260223_150000_a1b2c3d4")
            True
        """
        if not self.engine:
            return False
        ts = int(time.time())
        term = f"session('{session_id}', active, '', {ts})"
        try:
            self.engine.assertz(term)
            return True
        except Exception as e:
            logger.warning("Failed to assert session open %s: %s", session_id, e)
            return False

    @traces(
        spec="006",
        story="US-002",
        reqs=["REQ-006-FR5-001", "REQ-006-US-002-003", "REQ-006-US-002-006"],
    )
    def retract_session(self, session_id: str) -> bool:
        """Retract ``session(SessionID, _, _, _)`` from Prolog.

        Called when a session is closed to remove the active session fact.

        Args:
            session_id: The session identifier to retract.

        Returns:
            ``True`` if a fact was retracted, ``False`` otherwise.

        Example:
            >>> iface.retract_session("session_20260223_150000_a1b2c3d4")
            True
        """
        if not self.engine:
            return False
        try:
            results = list(
                self.engine.query(f"retract(session('{session_id}', _, _, _))")
            )
            return len(results) > 0
        except Exception as e:
            logger.warning("Failed to retract session %s: %s", session_id, e)
            return False

    @traces(spec="006", story="FR-008", reqs=["REQ-006-FR8-002"])
    def assert_session_fact(
        self, session_id: str, fact_term: str, source: str = ""
    ) -> bool:
        """Assert ``session_fact(SessionID, FactTerm, Timestamp, Source)``.

        Called when a fact is emitted. These facts accumulate in the
        knowledge base for the lifetime of the Prolog engine.

        Args:
            session_id: The owning session identifier.
            fact_term: A string describing the fact (e.g. a fact type or ID).
            source: Origin of the fact (e.g. ``"user"``, ``"system"``).

        Returns:
            ``True`` if asserted, ``False`` if no engine.

        Example:
            >>> iface.assert_session_fact("sess_001", "decision:use_redis", "architect")
            True
        """
        if not self.engine:
            return False
        ts = int(time.time())
        escaped_term = fact_term.replace("'", "\\'")
        escaped_source = source.replace("'", "\\'")
        term = (
            f"session_fact('{session_id}', '{escaped_term}', {ts}, '{escaped_source}')"
        )
        try:
            self.engine.assertz(term)
            return True
        except Exception as e:
            logger.warning(
                "Failed to assert session fact for session %s: %s", session_id, e
            )
            return False

    @traces(
        spec="006",
        story="US-001",
        reqs=[
            "REQ-006-FR2-003",
            "REQ-006-FR2-005",
            "REQ-006-FR9-001",
            "REQ-006-FR9-004",
            "REQ-006-US-001-006",
            "REQ-006-US-003-003",
            "REQ-006-US-003-005",
            "REQ-006-US-003-006",
        ],
    )
    def query_session(self, session_id: str) -> list[dict]:
        """Query the Prolog knowledge base for a specific session.

        Args:
            session_id: Session identifier to look up.

        Returns:
            List of solution dicts with keys ``Status``, ``Model`` and
            ``Timestamp`` (the canonical ``session/4`` predicate columns
            after ``Seq``).

        Example:
            >>> iface.query_session("sess_001")
            [{'Status': 'active', 'Model': '', 'Timestamp': 1740319200}]
        """
        if not self.engine:
            return []
        try:
            results = list(
                self.engine.query(f"session('{session_id}', Status, Model, Timestamp)")
            )
            return [dict(r) for r in results]
        except Exception as e:
            logger.warning("Failed to query session %s: %s", session_id, e)
            return []

    @traces(
        spec="006",
        story="US-003",
        reqs=["REQ-006-FR3-004", "REQ-006-FR9-002", "REQ-006-US-003-002"],
    )
    def query_session_facts(self, session_id: str) -> list[dict]:
        """Query all ``session_fact/4`` entries for a session.

        Args:
            session_id: Session identifier.

        Returns:
            List of dicts with keys ``FactTerm``, ``Timestamp``, ``Source``.
        """
        if not self.engine:
            return []
        try:
            results = list(
                self.engine.query(
                    f"session_fact('{session_id}', FactTerm, Timestamp, Source)"
                )
            )
            return [dict(r) for r in results]
        except Exception as e:
            logger.warning("Failed to query session facts for %s: %s", session_id, e)
            return []

    @traces(spec="006", story="US-004", reqs=["REQ-006-US-004-005"])
    def sync_facts(self) -> int:
        """Sync facts to engine.

        Returns:
            Number of facts asserted.
        """
        if not self.engine:
            return 0

        facts = self.fact_gen.generate_all_facts()
        self.facts_cache = facts

        count = 0
        for fact in facts:
            try:
                # Strip trailing period for the engine's assertz
                term = fact.rstrip(".")
                self.engine.assertz(term)
                count += 1
            except Exception as e:
                logger.warning("Failed to assert fact %r: %s", fact, e)

        return count

    @traces(spec="006", story="FR-005", reqs=["REQ-006-FR5-001"])
    def clear_facts(self) -> None:
        """Clear session facts from engine."""
        if not self.engine:
            return

        # Each predicate with its correct arity for retractall matching
        predicates = [
            ("project_current_session", 1),
            ("project_version", 1),
            ("project_branch", 1),
            ("project_status", 1),
            ("project_test_state", 1),
            ("session", 4),
            ("session_objective", 2),
            ("session_achievement", 2),
            ("session_artifact", 2),
            ("session_decision", 2),
            ("session_blocker", 2),
            ("session_active_work", 2),
            ("work_item", 5),
            ("work_item_notes", 2),
        ]

        for pred, arity in predicates:
            wildcards = ", ".join(["_"] * arity)
            try:
                self.engine.retractall(f"{pred}({wildcards})")
            except Exception as e:
                logger.warning("Failed to retractall %s/%d: %s", pred, arity, e)

    @traces(spec="006", story="FR-004", reqs=["REQ-006-FR4-003", "REQ-006-FR9-003"])
    def query_sessions(self, **criteria) -> list[dict]:
        """Query sessions.

        Args:
            **criteria: Query criteria.

        Returns:
            Matching sessions.
        """
        if not self.engine:
            return []

        results = []
        try:
            query = self._session_query(**criteria)
            results.extend(self.engine.query(query))
        except Exception as e:
            logger.warning("Failed to query sessions with criteria %s: %s", criteria, e)

        return results

    @traces(spec="006", story="FR-009", reqs=["REQ-006-FR9-003"])
    def query_work(self, **criteria) -> list[dict]:
        """Query work items.

        Args:
            **criteria: Query criteria.

        Returns:
            Matching work items.
        """
        if not self.engine:
            return []

        results = []
        try:
            query = self._work_query(**criteria)
            results.extend(self.engine.query(query))
        except Exception as e:
            logger.warning(
                "Failed to query work items with criteria %s: %s", criteria, e
            )

        return results

    @traces(
        spec="007",
        story="FR-010: Administrative Interface",
        reqs=["REQ-007-US-006-007"],
    )
    def export_facts(self, output_path: str | Path) -> int:
        """Export facts to file.

        Args:
            output_path: Output path.

        Returns:
            Number of facts written.
        """
        return self.fact_gen.write_facts_file(output_path)

    def reload_facts(self) -> int:
        """Reload facts from context.

        Returns:
            Number of facts loaded.
        """
        self.clear_facts()
        return self.sync_facts()

    def get_cached_facts(self) -> list[str]:
        """Get cached facts."""
        return self.facts_cache.copy()

    @traces(
        spec="006", story="US-001", reqs=["REQ-006-US-001-006", "REQ-006-US-002-006"]
    )
    def fact_exists(self, fact: str) -> bool:
        """Check if fact exists.

        Args:
            fact: Fact to check.

        Returns:
            True if exists.
        """
        if not self.engine:
            return False

        try:
            results = list(self.engine.query(fact))
            return len(results) > 0
        except Exception as e:
            logger.warning("Failed to check fact existence %r: %s", fact, e)
            return False

    @traces(spec="006", story="FR-002", reqs=["REQ-006-FR2-003", "REQ-006-FR9-003"])
    def execute_query(self, query: str) -> list[dict]:
        """Execute arbitrary query.

        Args:
            query: Prolog query.

        Returns:
            Query results.
        """
        if not self.engine:
            return []

        results = []
        try:
            results.extend(self.engine.query(query))
        except Exception as e:
            logger.warning("Failed to execute Prolog query %r: %s", query, e)

        return results

    @traces(
        spec="006",
        story="FR-004",
        reqs=["REQ-006-FR4-003", "REQ-006-FR9-003"],
    )
    def _session_query(self, **criteria) -> str:
        """Build session query."""
        parts = ["session(Seq, Purpose, Model, Timestamp)"]

        if "purpose_contains" in criteria:
            pattern = criteria["purpose_contains"]
            parts.append(f"sub_string(Purpose, _, _, _, '{pattern}')")

        if "seq" in criteria:
            parts[0] = parts[0].replace("Seq", str(criteria["seq"]))

        return ", ".join(parts)

    @traces(spec="006", story="FR-009", reqs=["REQ-006-FR9-003"])
    def _work_query(self, **criteria) -> str:
        """Build work query."""
        parts = ["work_item(Id, Text, State, Priority, Timestamp)"]

        if "state" in criteria:
            parts[0] = parts[0].replace("State", f"'{criteria['state']}'")

        if "priority" in criteria:
            parts[0] = parts[0].replace("Priority", f"'{criteria['priority']}'")

        return ", ".join(parts)

    def stats(self) -> dict:
        """Get interface statistics."""
        return {
            "engine_connected": self.engine is not None,
            "cached_facts": len(self.facts_cache),
            "root_path": str(self.root),
        }