orchestrata_ultimate.verification.constraints.evaluator

Reference › orchestrata_ultimate › verification › constraints › evaluator

ConstraintEvaluator is the mechanical judge of the verification subsystem. Its governing rule, quoted from its docstring, is SPEC-007 Key Principle 1: Prolog decisions are final. Every satisfaction decision is made by querying constraint_satisfied/3 in the constraints.pl rulebase through the prolog adapter; there is no Python comparison fallback. It is used by the verification agent, the checkpoint verifier and the verification engine.

A PrologEngine may be injected; otherwise _bridge builds one lazily on first use and consults the rulebase once, so constructing an evaluator is cheap. If no Prolog backend is installed, building the engine raises rather than degrading to a Python answer.

• evaluate looks up the actual value for the constraint's predicate in the context dictionary. A missing value gives an ERROR result; otherwise the constraint type, expected value and actual value are marshalled into ground Prolog terms (_to_prolog_term, with _escape_prolog_atom for atoms) and the goal is queried, giving SATISFIED or NOT_MET.
• evaluate_all maps evaluate over a list.
• evaluate_with_dependencies evaluates recursively in dependency order; a constraint whose dependency is unsatisfied is marked NOT_MET without being evaluated, and an unknown id yields an ERROR result.

Results are VerificationResult objects from types.

At a glance

Class diagram

Dependency diagram

API

Classes

ConstraintEvaluator

Decorators: traces(spec='035', story='US-003', reqs=['REQ-035-US-003-001', 'REQ-035-US-003-002', 'REQ-035-US-003-011']) · Line 71

Evaluates constraints against actual state via Prolog.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/constraints/evaluator.py

# mypy: ignore-errors
"""Constraint evaluation via Prolog."""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.paths import get_prolog_dir
from orchestrata_ultimate.core.prolog_adapter import PrologEngine
from orchestrata_ultimate.verification.types import (
    Constraint,
    VerificationResult,
    VerificationStatus,
)

logger = logging.getLogger(__name__)

_CONSTRAINTS_PL = "constraints.pl"


@traces(spec="035", story="US-003", reqs=["REQ-035-US-003-001"])
def _escape_prolog_atom(value: str) -> str:
    """Escape a string for safe use inside a single-quoted Prolog atom.

    Order matters: backslash MUST be escaped first, otherwise the
    backslashes introduced by the later replacements would themselves get
    re-escaped. Mirrors governance.prolog_gate._escape_prolog_atom -- each
    Prolog-marshalling module in this codebase carries its own copy rather
    than sharing a private helper across modules.
    """
    return (
        value.replace("\\", "\\\\")
        .replace("'", "\\'")
        .replace('"', '\\"')
        .replace("\n", "\\n")
    )


@traces(spec="035", story="US-003", reqs=["REQ-035-US-003-001"])
def _to_prolog_term(value: Any) -> str:
    """Marshal a Python value into a GROUND Prolog term literal.

    Only the shapes Constraint.expected/actual are documented to carry are
    supported: None, bool, int, float, str. ``bool`` is checked before
    ``int`` -- Python's bool is an int subclass, but constraints.pl treats
    true/false as distinct atoms from numbers (see truthy/1), so marshalling
    a bool as a number here would silently diverge from the rulebase.
    Anything else raises: producing a plausible-looking-but-wrong Prolog
    term for an unsupported type would be exactly the silent-fallback defect
    this evaluator exists to remove (constraint-prolog-evaluation.contract.json).
    """
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return repr(value)
    if isinstance(value, str):
        return f"'{_escape_prolog_atom(value)}'"
    raise TypeError(
        f"ConstraintEvaluator cannot marshal {type(value).__name__!r} to "
        "Prolog (supported: None, bool, int, float, str)"
    )


@traces(
    spec="035",
    story="US-003",
    reqs=["REQ-035-US-003-001", "REQ-035-US-003-002", "REQ-035-US-003-011"],
)
class ConstraintEvaluator:
    """Evaluates constraints against actual state via Prolog.

    SPEC-007 Key Principle #1: "Prolog decisions are final and cannot be
    overridden." Every satisfaction decision below is made by querying
    ``constraint_satisfied/3`` in prolog/constraints.pl -- there is no
    Python comparison ladder standing in as a fallback (backlog P2.10;
    plans/contracts/constraint-prolog-evaluation.contract.json).

    ``prolog_bridge`` is an optional injected ``PrologEngine`` (real
    janus-swi, e.g. the ``real_prolog`` test fixture or a production
    engine). When omitted, a real engine is constructed lazily on first
    use in :meth:`evaluate` -- dependency-injection convenience, not a
    decision fallback: every path still queries Prolog, never Python
    comparison logic. If no Prolog backend is available at all,
    constructing that engine raises (``core.prolog_adapter.PrologEngine``
    raises ``RuntimeError`` when janus-swi is not installed) rather than
    degrading to a Python-computed answer.
    """

    def __init__(self, prolog_bridge: PrologEngine | None = None) -> None:
        self._prolog = prolog_bridge
        self._rules_loaded = False

    @traces(
        spec="035", story="US-003", reqs=["REQ-035-US-003-001", "REQ-035-US-003-011"]
    )
    def _bridge(self) -> PrologEngine:
        """Return a working Prolog engine with constraints.pl consulted.

        Lazy on both counts: the engine is constructed (if not injected)
        and the rulebase is consulted only on first use, never in
        ``__init__`` -- so a bare ``ConstraintEvaluator()`` stays cheap and
        side-effect-free until a constraint is actually evaluated.
        """
        if self._prolog is None:
            self._prolog = PrologEngine()
        if not self._rules_loaded:
            self._prolog.consult(str(get_prolog_dir() / _CONSTRAINTS_PL))
            self._rules_loaded = True
        return self._prolog

    @traces(
        spec="035",
        story="US-003",
        reqs=["REQ-035-US-003-001", "REQ-035-US-003-005"],
    )
    def evaluate(
        self, constraint: Constraint, context: dict[str, Any] | None = None
    ) -> VerificationResult:
        """Evaluate a single constraint by querying constraint_satisfied/3."""
        ctx = context or {}
        actual = ctx.get(constraint.predicate)

        if actual is None:
            return VerificationResult(
                constraint_id=constraint.id,
                status=VerificationStatus.ERROR,
                expected=constraint.expected,
                actual=None,
                details=f"No value provided for '{constraint.predicate}'",
            )

        bridge = self._bridge()
        goal = (
            "constraint_satisfied("
            f"{_to_prolog_term(str(constraint.type))}, "
            f"{_to_prolog_term(constraint.expected)}, "
            f"{_to_prolog_term(actual)})"
        )
        # PrologEngine.query_once always returns a dict (janus's {'truth':
        # False} stripped down to {}) -- it is never None for an ordinary
        # failing goal, so "is not None" cannot distinguish satisfied from
        # not-met. list(bridge.query(goal)) is the established idiom for a
        # yes/no query in this codebase (governance.prolog_gate,
        # kernel.policy_evaluator): [] means the goal failed.
        satisfied = bool(list(bridge.query(goal)))

        return VerificationResult(
            constraint_id=constraint.id,
            status=(
                VerificationStatus.SATISFIED
                if satisfied
                else VerificationStatus.NOT_MET
            ),
            expected=constraint.expected,
            actual=actual,
            details=f"{'Satisfied' if satisfied else 'Not met'}: {constraint.predicate}",
        )

    @traces(
        spec="035",
        story="US-003",
        reqs=[
            "REQ-035-US-003-002",
            "REQ-035-US-003-007",
            "REQ-035-US-003-012",
            "REQ-035-US-003-020",
        ],
    )
    def evaluate_all(
        self, constraints: list[Constraint], context: dict[str, Any] | None = None
    ) -> list[VerificationResult]:
        """Evaluate all constraints."""
        return [self.evaluate(c, context) for c in constraints]

    @traces(
        spec="035",
        story="US-002",
        reqs=[
            "REQ-035-US-002-006",
            "REQ-035-US-002-013",
            "REQ-035-US-002-021",
            "REQ-035-US-003-004",
        ],
    )
    def evaluate_with_dependencies(
        self, constraints: list[Constraint], context: dict[str, Any] | None = None
    ) -> list[VerificationResult]:
        """Evaluate constraints respecting dependency order."""
        evaluated: dict[str, VerificationResult] = {}
        constraint_map = {c.id: c for c in constraints}

        @traces(
            spec="035",
            story="US-002",
            reqs=[
                "REQ-035-US-002-006",
                "REQ-035-US-002-013",
                "REQ-035-US-002-021",
                "REQ-035-US-003-004",
            ],
        )
        def eval_one(cid: str) -> VerificationResult:
            if cid in evaluated:
                return evaluated[cid]
            constraint = constraint_map.get(cid)
            if not constraint:
                result = VerificationResult(
                    constraint_id=cid,
                    status=VerificationStatus.ERROR,
                    details=f"Constraint '{cid}' not found",
                )
                evaluated[cid] = result
                return result
            for dep_id in constraint.dependencies:
                dep_result = eval_one(dep_id)
                if not dep_result.is_satisfied:
                    result = VerificationResult(
                        constraint_id=cid,
                        status=VerificationStatus.NOT_MET,
                        details=f"Dependency '{dep_id}' not satisfied",
                    )
                    evaluated[cid] = result
                    return result
            result = self.evaluate(constraint, context)
            evaluated[cid] = result
            return result

        for c in constraints:
            eval_one(c.id)
        return list(evaluated.values())