orchestrata_ultimate.core.prolog_adapter

Reference › orchestrata_ultimate › core › prolog_adapter

The Prolog engine adapter over janus-swi, in core. It imports janus through janus_bootstrap and sets JANUS_AVAILABLE. If janus is not importable, constructing a PrologEngine raises RuntimeError; there is no in-memory or alternative backend.

PrologEngine pins a reusable SWI engine to the calling thread (_ensure_thread_engine), releases it in close (also used by the context-manager protocol), and exposes consult, assertz, retract, retractall, query and query_once. Janus returns bindings with a truth key, which _strip_truth removes so callers see plain binding dictionaries. query_explained runs a goal through the explain.pl meta-interpreter of ADR-0031, loading it once per process, and returns a ProofResult with the success flag, the proof tree and a list of why-not failures.

One limitation shapes usage: assertz goes through a goal string, and janus treats capitalised names there as Python-bound inputs, so asserting a clause with Prolog variables fails with an instantiation error. Variable-bearing rules therefore belong in .pl files loaded with consult, and only ground facts are asserted at runtime; the governance gate follows this pattern with its completion rules.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ProofResult (dataclass)

Decorators: dataclass(frozen=True) · Line 62

The outcome of an explained Prolog query (ADR-0031 / SPEC-033).

PrologEngine

Line 77

Prolog engine over janus-swi.

Functions

Source

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

# mypy: ignore-errors
"""Prolog engine adapter -- janus-swi.

Provides a unified PrologEngine class over janus-swi (thread-safe). janus
returns query results as ``{'truth': True, 'X': val}``; the adapter strips the
``'truth'`` key so callers get plain ``{'X': val}`` binding dicts.

If janus-swi is not available, constructing a PrologEngine raises RuntimeError --
there is no in-memory fallback and no other backend.

LIMITATION + the pattern that works around it:
``assertz`` goes through ``query_once("assertz(<term>)")``, and janus treats
capitalised names in a goal string as Python-bound *input* variables -- so
asserting a clause that contains Prolog **variables** (e.g. ``foo(X) :- bar(X)``)
raises "Arguments are not sufficiently instantiated". Ground facts (no variables)
assert fine.

So the rule is: keep variable-bearing *rules* in a ``.pl`` file and load them
with ``consult`` (which parses Prolog source natively, variables and all), and
only ``assertz`` **ground** facts at runtime. ``governance.PrologGate`` follows
exactly this pattern -- its ``may_complete`` rules live in ``prolog/completion.pl``
(consulted) while task facts are asserted ground.
"""

import logging
import threading
from dataclasses import dataclass, field
from typing import Any

logger = logging.getLogger(__name__)

JANUS_AVAILABLE = False
# Typed Any: janus_swi ships no stubs, and the None initializer would otherwise
# make pyright reject every `_janus.<attr>` access as Optional-member / unknown.
_janus: Any = None

try:
    from orchestrata_ultimate.core.janus_bootstrap import (
        janus as _janus,  # type: ignore[no-redef]
    )

    JANUS_AVAILABLE = True
    logger.info("Prolog backend: janus-swi (thread-safe)")
except ImportError as exc:
    logger.debug("janus-swi not available: %s", exc)

PROLOG_AVAILABLE = JANUS_AVAILABLE

# ADR-0031 / SPEC-033: explain.pl is consulted at most once per process (the
# janus engine is process-global -- see tests/conftest.py's real_prolog note)
# the first time query_explained is used, guarded by _EXPLAIN_LOCK.
_EXPLAIN_LOADED = False
_EXPLAIN_LOCK = threading.Lock()


def _strip_truth(result: dict[str, Any]) -> dict[str, Any]:
    """Strip janus's 'truth' key, leaving a plain binding dict."""
    return {k: v for k, v in result.items() if k != "truth"}


@dataclass(frozen=True)
class ProofResult:
    """The outcome of an explained Prolog query (ADR-0031 / SPEC-033).

    ``tree`` (success) and ``whynot`` (failure) are always JSON-safe -- plain
    dicts/lists/atoms/numbers -- because that is the only shape janus can
    marshal out of a query (see explain.pl's MARSHALING note); there is no
    separate conversion step on the Python side.
    """

    succeeded: bool
    tree: dict[str, Any] | None
    whynot: list[dict[str, Any]] = field(default_factory=list)
    goal: str = ""


class PrologEngine:
    """Prolog engine over janus-swi.

    Concurrency & performance model (measured, not assumed):
    janus-swi already runs every op on the calling thread's OWN SWI engine --
    ``swipl_call`` / ``swipl_open_query`` do ``PL_thread_attach_engine`` /
    ``PL_thread_destroy_engine`` internally -- and it releases the GIL around
    SWI's solution search. So concurrent governance queries are correct AND
    genuinely parallel on stock janus with no help from this class; that is not
    what the per-thread attach below buys.

    What it buys is engine REUSE. Left to itself janus creates and destroys a
    fresh SWI engine on EVERY call (``janus.engine()`` reads -1 between ops).
    Attaching a persistent engine once per thread pins it (a stable id), so
    subsequent ops reuse it instead of paying the create/destroy churn --
    measured ~3.4x higher throughput on a hot concurrent governance-query path.
    This is a latency optimization, not a correctness requirement.

    Use ``close()`` / the context manager to release the calling thread's
    pinned engine.
    """

    def __init__(self, persistent: bool = True) -> None:
        """Initialize the Prolog engine.

        Args:
            persistent: If True (default), pin a reusable Prolog engine on the
                constructing thread (and lazily on any other thread that later
                calls an op), avoiding per-call engine create/destroy churn.
        Raises:
            RuntimeError: If janus-swi is not available.
        """
        if not JANUS_AVAILABLE:
            raise RuntimeError(
                "No Prolog backend available. Install janus-swi (uv add janus-swi)."
            )
        self._backend = "janus"
        self._persistent = persistent
        self._engine = None
        # Per-(instance, thread) flag: has THIS instance pinned an engine on the
        # calling thread yet? Keeps the (idempotent, ref-counted) attach off the
        # hot path after first use, and -- critically -- lets close() release
        # ONLY threads this instance actually pinned (see close()).
        self._thread_state = threading.local()
        if persistent:
            self._ensure_thread_engine()

    def _ensure_thread_engine(self) -> None:
        """Pin a reusable SWI engine on the calling thread (once per thread)."""
        if not self._persistent or getattr(self._thread_state, "attached", False):
            return
        _janus.attach_engine()
        self._thread_state.attached = True

    def close(self) -> None:
        """Release the pinned engine on the CALLING thread, if this instance pinned one.

        ``detach_engine()`` acts on whichever thread runs it, so it must only be
        called on a thread this instance actually attached (tracked by the
        thread-local flag). Running it on an unrelated thread -- e.g. a
        finalizer thread during ``__del__`` -- would decrement/destroy an engine
        this instance never created. Engines pinned on other threads are
        reclaimed by SWI when those threads exit (bounded, long-lived pools).
        """
        if getattr(self._thread_state, "attached", False):
            try:
                _janus.detach_engine()
            except Exception:
                pass
            self._thread_state.attached = False

    def __del__(self) -> None:
        self.close()

    def __enter__(self) -> PrologEngine:
        return self

    def __exit__(self, *args: object) -> None:
        self.close()

    @property
    def backend(self) -> str:
        return self._backend

    def consult(self, path: str) -> None:
        self._ensure_thread_engine()
        _janus.consult(str(path))

    def assertz(self, term: str) -> None:
        # janus binds capitalised goal-string names as Python inputs, so this
        # works for GROUND facts only; variable-bearing clauses must be loaded
        # via consult of a .pl file (see module docstring).
        self._ensure_thread_engine()
        _janus.query_once(f"assertz({term})")

    def retract(self, term: str) -> None:
        self._ensure_thread_engine()
        _janus.query_once(f"retract({term})")

    def retractall(self, term: str) -> None:
        self._ensure_thread_engine()
        _janus.query_once(f"retractall({term})")

    def query(self, goal: str) -> list[dict[str, Any]]:
        self._ensure_thread_engine()
        return [_strip_truth(r) for r in _janus.query(goal)]

    def query_once(self, goal: str) -> dict[str, Any] | None:
        self._ensure_thread_engine()
        result = _janus.query_once(goal)
        return _strip_truth(result) if result is not None else None

    def query_explained(
        self, goal: str, bound: int = 50, **bindings: str
    ) -> ProofResult:
        """Execute ``goal`` through the ADR-0031 explain.pl meta-interpreter.

        Opt-in explaining query mode: this is a peer of ``query``/``query_once``,
        not a replacement -- the fast, non-explaining path those two methods
        serve is unaffected. Thin wrapper around the module-level
        ``query_explained`` (below) that additionally pins this instance's
        thread engine first, matching every other method on this class.
        """
        self._ensure_thread_engine()
        return query_explained(goal, bound, **bindings)


def _ensure_explain_loaded() -> None:
    """Consult explain.pl into the (process-global) janus engine, once."""
    global _EXPLAIN_LOADED
    if _EXPLAIN_LOADED:
        return
    with _EXPLAIN_LOCK:
        if _EXPLAIN_LOADED:
            return
        from orchestrata_ultimate.core.paths import get_prolog_dir

        _janus.consult(str(get_prolog_dir() / "explain.pl"))
        _EXPLAIN_LOADED = True


def query_explained(goal: str, bound: int = 50, **bindings: str) -> ProofResult:
    """Execute ``goal`` through the ADR-0031 explain.pl meta-interpreter.

    Module-level so both ``PrologEngine`` (above) and the separate
    ``PrologWorker``/``PrologProxy`` janus seam (orchestrator/prolog_worker.py
    -- the one ``PrologBridge``/``GovernancePrologClient`` actually use in
    production) can call the SAME logic against the SAME process-global janus
    module, instead of a duplicated (and driftable) second implementation.

    ``goal`` may contain ``{name}`` placeholders, resolved via
    ``str.format(**bindings)`` before it is sent to Prolog (bindings must
    already be Prolog-atom-safe strings, same as every other query-string
    builder in this codebase).

    Returns a ``ProofResult``: on success, ``tree`` names the rules fired and
    facts consumed (in order); on failure, ``whynot`` enumerates each
    candidate clause's first failing precondition, bounded by ``bound`` so a
    recursive/cyclic precondition cannot hang the call.
    """
    _ensure_explain_loaded()
    resolved_goal = goal.format(**bindings) if bindings else goal

    tree = _try_explain_call(resolved_goal)
    if tree is not None:
        return ProofResult(succeeded=True, tree=tree, whynot=[], goal=resolved_goal)

    whynot = _try_explain_whynot(resolved_goal, bound)
    return ProofResult(succeeded=False, tree=None, whynot=whynot, goal=resolved_goal)


def _try_explain_call(resolved_goal: str) -> dict[str, Any] | None:
    """Run explain_call/2; return its Tree dict, or None if Goal failed.

    NOTE: janus_swi's query_once/1 ALWAYS returns a dict carrying a `truth`
    key (True/False) -- it never returns Python None on failure (verified
    directly against this repo's janus_swi build) -- so success must be read
    from `truth`, not from an `is not None` check on the raw result.
    """
    try:
        solved = _janus.query_once(f"explain:explain_call(({resolved_goal}), Tree)")
    except Exception as exc:
        logger.debug("explain_call failed for %r: %s", resolved_goal, exc)
        return None
    if solved is not None and solved.get("truth"):
        return solved["Tree"]
    return None


def _try_explain_whynot(resolved_goal: str, bound: int) -> list[dict[str, Any]]:
    """Run explain_whynot/3; return its Failures list (possibly empty)."""
    try:
        explained = _janus.query_once(
            f"explain:explain_whynot(({resolved_goal}), {bound}, Failures)"
        )
    except Exception as exc:
        logger.debug("explain_whynot failed for %r: %s", resolved_goal, exc)
        return []
    if explained is not None and explained.get("truth"):
        return list(explained["Failures"])
    return []