orchestrata_ultimate.orchestrator.prolog_worker

Reference › orchestrata_ultimate › orchestrator › prolog_worker

The gateway to the janus-swi Prolog engine for governance queries. It is imported by the PrologBridge implementations in infra.prolog_bridge and cxp.prolog_bridge, so the rule evaluation behind engine, intent_validator and intent_router runs through it. It relies on core.janus_bootstrap and core.prolog_adapter.

PrologWorker.start imports janus lazily, so the module can be imported without loading the SWI library, and raises RuntimeError when janus-swi is unavailable. submit takes a PrologOp (consult, query, assertz or retract) and a payload string, runs it with _execute and returns a Future holding the result or the exception; it raises RuntimeError if the worker is not alive. _ensure_thread_engine pins a reusable SWI engine to each calling thread once, which the docstring says speeds up the concurrent path. query_explained delegates to the explaining query in core.prolog_adapter (ADR-0031). PrologProxy presents the consult, query, assertz and retract interface of the earlier pyswip binding on top of the worker, with a 30-second wait on results. get_prolog_worker returns a process-wide singleton created under a lock on first use and recreated if it is no longer alive, and get_prolog_proxy wraps it.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PrologOp (enum)

Bases: Enum · Line 43

Operations the Prolog worker can execute.

PrologRequest (dataclass)

Decorators: dataclass · Line 54

A request submitted to the Prolog worker.

PrologWorker

Decorators: traces(spec='007', story='US-PGOV-001', reqs=['REQ-007-US-001-005']) · Line 73

Janus-swi Prolog worker.

PrologProxy

Line 231

Drop-in replacement for pyswip.Prolog that routes through janus-swi.

Functions

Source

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

# mypy: ignore-errors
"""Janus-swi Prolog worker for thread-safe, PARALLEL governance queries.

Janus-swi is thread-safe, so operations execute directly in the calling
thread without a dedicated worker thread. This replaces the pyswip-based
queue worker that required single-threaded access.

Architecture:
    Thread-1 --+   (own SWI engine, pinned for reuse)
    Thread-2 --+-- submit(op, payload) --> janus_swi.query / assertz / consult
    Thread-3 --+   (own SWI engine, pinned for reuse)

Parallelism -- native to janus, NOT provided by this worker:
    janus-swi runs every op on the calling thread's own SWI engine (its
    ``swipl_call`` / ``swipl_open_query`` attach a per-thread engine) and
    RELEASES the GIL around ``PL_next_solution`` (SWI's solution search). So
    governance goals from different threads already run CONCURRENTLY and
    correctly on stock janus -- ~2.6x on 8 threads for CPU-bound recursion --
    with no help from this class. That also makes the free-threaded (GIL-off)
    janus patch UNNECESSARY, and stock janus stays tabling-safe.

What ``_ensure_thread_engine`` DOES buy is engine REUSE, a latency win:
    Left alone, janus creates and DESTROYS a fresh engine on every call
    (``janus.engine()`` reads -1 between ops). Pinning an engine once per
    thread lets subsequent ops reuse it instead of paying create/destroy
    churn -- measured ~3.4x higher throughput on a hot concurrent query path.
    The clause database is shared process-wide, so predicates consulted on any
    thread are visible to every engine.
"""

import logging
import threading
from concurrent.futures import Future
from dataclasses import dataclass
from enum import Enum, auto
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

_logger = logging.getLogger(__name__)


class PrologOp(Enum):
    """Operations the Prolog worker can execute."""

    CONSULT = auto()
    QUERY = auto()
    ASSERTZ = auto()
    RETRACT = auto()
    SHUTDOWN = auto()


@dataclass
class PrologRequest:
    """A request submitted to the Prolog worker.

    Attributes:
        op: The Prolog operation to perform.
        payload: The Prolog query/fact/file string.
        future: Future that receives the result.
    """

    op: PrologOp
    payload: str
    future: Future


@traces(
    spec="007",
    story="US-PGOV-001",
    reqs=["REQ-007-US-001-005"],
)
class PrologWorker:
    """Janus-swi Prolog worker.

    All Prolog operations are delegated directly to janus_swi,
    which is thread-safe and does not require a dedicated worker
    thread.
    """

    def __init__(self) -> None:
        """Initialize the worker."""
        self._alive = False
        self._janus: Any = None
        # Tracks which threads have pinned a reusable SWI engine, so the
        # (idempotent, ref-counted) attach is a cheap flag check on the hot
        # path instead of a C call per op.
        self._engine_state = threading.local()

    def start(self) -> None:
        """Initialize janus-swi."""
        if self._alive:
            return
        try:
            from orchestrata_ultimate.core.janus_bootstrap import (
                janus,  # type: ignore[no-redef]
            )

            self._janus = janus
            self._alive = True
            _logger.info("Prolog worker started (janus-swi)")
        except ImportError as exc:
            _logger.error("janus-swi not available: %s", exc)
            raise RuntimeError("janus-swi is required") from exc

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-005"])
    def _ensure_thread_engine(self) -> None:
        """Pin a reusable SWI engine on this thread (once) to avoid churn.

        janus already runs each op on the thread's own engine, but destroys it
        after every call; pinning one keeps it alive so subsequent ops reuse it
        (~3.4x on a hot concurrent path). ``attach_engine`` is idempotent
        ("if needed"); the thread-local flag keeps it off the hot path after
        the first call. Pinned engines live for the thread's lifetime -- the
        worker pool is bounded and the process is long-lived, so there is
        nothing to detach.
        """
        if getattr(self._engine_state, "attached", False):
            return
        attach = getattr(self._janus, "attach_engine", None)
        if attach is not None:
            attach()
        self._engine_state.attached = True

    def _execute(self, req: PrologRequest) -> Any:
        """Execute a single Prolog operation via janus-swi.

        Args:
            req: The request to execute.

        Returns:
            Operation-specific result.
        """
        janus = self._janus
        self._ensure_thread_engine()
        if req.op == PrologOp.CONSULT:
            janus.consult(req.payload)
            return True
        elif req.op == PrologOp.QUERY:
            return [_strip_truth(r) for r in janus.query(req.payload)]
        elif req.op == PrologOp.ASSERTZ:
            janus.query_once(f"assertz({req.payload})")
            return True
        elif req.op == PrologOp.RETRACT:
            janus.query_once(f"retract({req.payload})")
            return True
        return None

    def submit(self, op: PrologOp, payload: str) -> Future:
        """Submit a Prolog operation.

        Args:
            op: The operation to perform.
            payload: The Prolog query/fact/file string.

        Returns:
            A Future that resolves with the operation result.

        Raises:
            RuntimeError: If the worker is not alive.
        """
        if not self._alive:
            raise RuntimeError("Prolog worker is not running")
        future: Future = Future()
        try:
            result = self._execute(PrologRequest(op=op, payload=payload, future=future))
            future.set_result(result)
        except Exception as exc:
            future.set_exception(exc)
        return future

    @property
    def is_alive(self) -> bool:
        """Whether the worker is running."""
        return self._alive

    def shutdown(self) -> None:
        """No-op shutdown — janus-swi is global and lives for the process."""
        self._alive = False
        _logger.info("Prolog worker shut down")

    @traces(
        spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-007", "REQ-007-FR10-032"]
    )
    def query_explained(
        self, goal: str, bound: int = 50, **bindings: str
    ) -> Any:  # ProofResult — Any: importing core.prolog_adapter for the
        # annotation loads janus at import time (deliberately deferred to
        # start()), and TYPE_CHECKING-only names NameError under PEP 649/749.
        """Opt-in explaining query (ADR-0031) -- see
        core.prolog_adapter.query_explained.

        Delegates to the module-level implementation shared with
        ``PrologEngine`` so this worker does not carry a second copy of the
        explain.pl querying logic; only the thread-engine pinning (this
        worker's own optimization) is worker-specific.

        The import is deliberately LOCAL: ``core.prolog_adapter`` loads
        janus (the SWI shared library) at ITS import time, and this module
        must stay importable without that side effect -- ``start()`` imports
        janus lazily for exactly the same reason.
        """
        from orchestrata_ultimate.core.prolog_adapter import (
            query_explained as _query_explained,
        )

        if not self._alive:
            raise RuntimeError("Prolog worker is not running")
        self._ensure_thread_engine()
        return _query_explained(goal, bound, **bindings)


# ============================================================================
# Module-level singleton — shared by all PrologBridge instances
# ============================================================================

_WORKER: PrologWorker | None = None
_WORKER_LOCK: Any = None


@traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-005"])
def _get_lock() -> Any:
    global _WORKER_LOCK
    if _WORKER_LOCK is None:
        import threading

        _WORKER_LOCK = threading.Lock()
    return _WORKER_LOCK


class PrologProxy:
    """Drop-in replacement for pyswip.Prolog that routes through janus-swi.

        Exposes the same consult/query/assertz/retract interface so existing
    code (like the CXP PrologBridge) can use it with zero refactoring.

        Usage:
            proxy = get_prolog_proxy()
            proxy.consult("/path/to/rules.pl")
            results = list(proxy.query("member(X, [1,2,3])"))
    """

    def __init__(self, worker: PrologWorker) -> None:
        """Initialize proxy backed by a PrologWorker.

        Args:
            worker: The PrologWorker to delegate to.
        """
        self._worker = worker

    def consult(self, path: str) -> None:
        """Load a Prolog file.

        Args:
            path: Absolute path to the .pl file.
        """
        self._worker.submit(PrologOp.CONSULT, path).result(timeout=30)

    def query(self, query_str: str) -> list[dict[str, Any]]:
        """Execute a Prolog query.

        Args:
            query_str: The Prolog query string.

        Returns:
            List of solution dictionaries.
        """
        return self._worker.submit(PrologOp.QUERY, query_str).result(timeout=30)

    def assertz(self, fact_str: str) -> None:
        """Assert a fact at the end of the database.

        Args:
            fact_str: The Prolog fact string.
        """
        self._worker.submit(PrologOp.ASSERTZ, fact_str).result(timeout=10)

    def retract(self, fact_str: str) -> None:
        """Retract a fact from the database.

        Args:
            fact_str: The Prolog fact string.
        """
        self._worker.submit(PrologOp.RETRACT, fact_str).result(timeout=10)

    @traces(
        spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-007", "REQ-007-FR10-032"]
    )
    def query_explained(
        self, goal: str, bound: int = 50, **bindings: str
    ) -> Any:  # ProofResult — Any: importing core.prolog_adapter for the
        # annotation loads janus at import time (deliberately deferred to
        # start()), and TYPE_CHECKING-only names NameError under PEP 649/749.
        """Opt-in explaining query (ADR-0031) -- see
        core.prolog_adapter.query_explained.

        Not routed through ``submit``/``PrologOp`` (that path is a single
        string payload; explanation needs a goal, a bound, and keyword
        bindings), so this calls the worker directly -- which, like every
        other op here, executes synchronously on the calling thread.
        """
        return self._worker.query_explained(goal, bound, **bindings)


def _strip_truth(result: dict[str, Any]) -> dict[str, Any]:
    """Remove janus 'truth' key for pyswip-compatible output."""
    return {k: v for k, v in result.items() if k != "truth"}


@traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-005"])
def get_prolog_worker() -> PrologWorker:
    """Get or create the singleton PrologWorker.

    Thread-safe via double-checked locking. The worker is created
    on first call, then reused for all subsequent calls.

    Returns:
        The running PrologWorker singleton.
    """
    global _WORKER
    if _WORKER is None or not _WORKER.is_alive:
        with _get_lock():
            if _WORKER is None or not _WORKER.is_alive:
                _WORKER = PrologWorker()
                _WORKER.start()
    return _WORKER


def get_prolog_proxy() -> PrologProxy:
    """Get a PrologProxy backed by the singleton worker.

    Drop-in replacement for pyswip.Prolog — same API but thread-safe
    via janus-swi. Use this in place of Prolog() to ensure all
    Prolog access goes through janus.

    Returns:
        A PrologProxy instance.
    """
    return PrologProxy(get_prolog_worker())