orchestrata_ultimate.factory.anchors.evaluation

Reference › orchestrata_ultimate › factory › anchors › evaluation

evaluation is the satisfaction-checking half of the semantic-anchor subsystem (SPEC-033 row B15b, ADR-0013 and ADR-0023). The models and store modules author and keep anchors; this module compares a ratified Anchor with an AnchorEvidence (the outputs the artifact under test already produced, plus a differential map of prior behaviour) and returns a Satisfaction: SATISFIED, UNSATISFIED or INCONCLUSIVE. It makes no model call and executes no code. Its importers are the gate, the fast-path classifier and the scripts.factory_anchors and scripts.factory_fastpath CLIs.

There is one evaluator per AnchorClass, each satisfying the AnchorEvaluator protocol (a klass property and an evaluate method that returns an AnchorResult):

• ExampleEvaluator parses a statement of the form given: ... | expect: key=value and compares the observed value for that key.
• PropertyEvaluator parses an and-joined predicate of clauses var OP literal, where the operator is one of ==, !=, <, <=, >, >= or in. A hand-written tokenizer (_parse_property_clause, _parse_scalar, _parse_literal) does the parsing and eval and exec are never used. A malformed statement raises PropertyParseError. A variable missing from the evidence makes the result INCONCLUSIVE, as _missing_vars reports.
• OracleEvaluator checks that observed outputs equal the prior behaviour in differential on the observed keys.

evaluate_all dispatches a batch of anchors to the evaluator matching each anchor's class. An anchor whose class has no evaluator is INCONCLUSIVE and is never skipped silently.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

Satisfaction (enum)

Bases: StrEnum · Line 42

The three outcomes of comparing an anchor against observed evidence.

AnchorResult (dataclass)

Decorators: dataclass(frozen=True) · Line 51

The outcome of evaluating one anchor against one :class:AnchorEvidence.

AnchorEvidence (dataclass)

Decorators: dataclass(frozen=True) · Line 61

Already-observed outputs an evaluator compares an anchor against.

AnchorEvaluator (protocol)

Bases: Protocol · Line 86

Anything that can decide satisfaction for one :class:~.models.AnchorClass.

ExampleEvaluator

Line 133

Evaluates an EXAMPLE anchor: an observed key must equal its expected value.

PropertyParseError

Bases: ValueError · Line 179

A property statement does not match the restricted grammar.

PropertyEvaluator

Line 309

Evaluates a PROPERTY anchor: an and-joined predicate over outputs.

OracleEvaluator

Line 368

Evaluates a DIFFERENTIAL_ORACLE anchor: behaviour must be unchanged vs prior.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/anchors/evaluation.py

"""Anchor evaluation — satisfaction checking (B15b, ADR-0013/0023).

B15a built anchor AUTHORING + STORAGE. This module is the pure comparison
step: given a ratified :class:`~.models.Anchor` and
:class:`AnchorEvidence` (outputs the artifact-under-test ALREADY produced),
decide :class:`Satisfaction`. There is **no model call and no code
execution** here — running the artifact and collecting its outputs is the
harness's job (out of scope for B15b); this module only compares already-
observed values.

Three real evaluators, one per :class:`~.models.AnchorClass`:

- :class:`ExampleEvaluator` — parses the anchor's ``given: ... | expect:
  <key>=<value>`` statement (the shape :mod:`~orchestrata_ultimate.factory.
  dor_procedure.anchors_export` and :mod:`~.intake` produce/consume) and
  compares ``observed_outputs[key]`` against the expected value.
- :class:`PropertyEvaluator` — parses a small ``and``-joined predicate
  (``var OP literal``, ``OP`` in ``== != < <= > >= in``) with a
  hand-written tokenizer — **never** ``eval``/``exec`` — and checks it
  against ``observed_outputs``.
- :class:`OracleEvaluator` — compares ``observed_outputs`` against
  ``differential`` (the prior, observed behaviour) on the observed keys.

:func:`evaluate_all` dispatches a batch of anchors to the matching
evaluator by :class:`~.models.AnchorClass`; an anchor whose class has no
evaluator is INCONCLUSIVE, never silently skipped.
"""

from __future__ import annotations

import operator
import re
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from enum import StrEnum
from typing import Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.anchors.models import Anchor, AnchorClass


class Satisfaction(StrEnum):
    """The three outcomes of comparing an anchor against observed evidence."""

    SATISFIED = "satisfied"
    UNSATISFIED = "unsatisfied"
    INCONCLUSIVE = "inconclusive"


@dataclass(frozen=True)
class AnchorResult:
    """The outcome of evaluating one anchor against one :class:`AnchorEvidence`."""

    anchor_id: str
    klass: AnchorClass
    satisfaction: Satisfaction
    detail: str


@dataclass(frozen=True)
class AnchorEvidence:
    """Already-observed outputs an evaluator compares an anchor against.

    ``observed_outputs`` maps an AC-observable name to the value the
    artifact-under-test produced for it. ``differential`` maps the same
    kind of name to the PRIOR (pre-change) observed value, populated only
    when evaluating an amendment (ADR-0016) — empty for a fresh build.
    """

    observed_outputs: dict[str, str]
    differential: dict[str, str]


def _result(
    anchor: Anchor, klass: AnchorClass, satisfaction: Satisfaction, detail: str
) -> AnchorResult:
    """Build an :class:`AnchorResult` for ``anchor`` — cuts boilerplate below."""
    return AnchorResult(
        anchor_id=anchor.anchor_id,
        klass=klass,
        satisfaction=satisfaction,
        detail=detail,
    )


class AnchorEvaluator(Protocol):
    """Anything that can decide satisfaction for one :class:`~.models.AnchorClass`."""

    @property
    def klass(self) -> AnchorClass:
        """The anchor class this evaluator handles."""
        ...  # pragma: no cover -- Protocol stub, never invoked directly

    def evaluate(self, anchor: Anchor, evidence: AnchorEvidence) -> AnchorResult:
        """Compare ``anchor`` against ``evidence`` and return the verdict."""
        ...  # pragma: no cover -- Protocol stub, never invoked directly


# ---------------------------------------------------------------------------
# ExampleEvaluator — "given: ... | expect: <key>=<value>"
# ---------------------------------------------------------------------------


def _parse_example_statement(statement: str) -> tuple[str, str] | None:
    """Parse an EXAMPLE anchor statement into ``(observable_key, expected_value)``.

    Statement shape (see :func:`~orchestrata_ultimate.factory.anchors.intake.
    import_dor_candidates` and ADR-0012): ``"given: <precondition> | expect:
    <key>=<value>"`` (a ``"when: ..."`` clause may sit between given and
    expect; only the ``expect:`` segment is parsed here). Returns ``None``
    if no ``expect:`` segment or no parseable ``key=value`` pair is found —
    callers must treat that as INCONCLUSIVE, never guess.
    """
    expect_segment: str | None = None
    for segment in statement.split("|"):
        stripped = segment.strip()
        if stripped.lower().startswith("expect:"):
            expect_segment = stripped[len("expect:") :].strip()
            break
    if expect_segment is None:
        return None
    for sep in ("==", "="):
        if sep in expect_segment:
            key, _, value = expect_segment.partition(sep)
            key = key.strip()
            value = value.strip()
            if key and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_.]*", key):
                return key, value
            return None
    return None


class ExampleEvaluator:
    """Evaluates an EXAMPLE anchor: an observed key must equal its expected value."""

    @property
    def klass(self) -> AnchorClass:
        return AnchorClass.EXAMPLE

    def evaluate(self, anchor: Anchor, evidence: AnchorEvidence) -> AnchorResult:
        parsed = _parse_example_statement(anchor.statement)
        if parsed is None:
            return _result(
                anchor,
                self.klass,
                Satisfaction.INCONCLUSIVE,
                "anchor statement has no parseable 'expect: key=value' clause: "
                f"{anchor.statement!r}",
            )
        key, expected_value = parsed
        if key not in evidence.observed_outputs:
            return _result(
                anchor,
                self.klass,
                Satisfaction.INCONCLUSIVE,
                f"observed_outputs has no key {key!r}",
            )
        observed_value = evidence.observed_outputs[key]
        if observed_value == expected_value:
            return _result(
                anchor,
                self.klass,
                Satisfaction.SATISFIED,
                f"{key}={observed_value!r} matches expected {expected_value!r}",
            )
        return _result(
            anchor,
            self.klass,
            Satisfaction.UNSATISFIED,
            f"{key}: observed {observed_value!r} != expected {expected_value!r}",
        )


# ---------------------------------------------------------------------------
# PropertyEvaluator — "var OP literal (and var OP literal)*", hand-tokenized
# ---------------------------------------------------------------------------


class PropertyParseError(ValueError):
    """A property statement does not match the restricted grammar.

    Deliberately NOT a place where ``eval``/``exec`` ever runs — a statement
    that looks like it could invoke arbitrary code (e.g. contains
    ``__import__(...)``) fails to match the grammar and raises this instead
    of being executed.
    """


_PROPERTY_AND_SPLIT_RE = re.compile(r"\s+and\s+", re.IGNORECASE)
_PROPERTY_CLAUSE_RE = re.compile(
    r"^(?P<var>[A-Za-z_][A-Za-z0-9_.]*)\s*"
    r"(?P<op>==|!=|<=|>=|<|>|in)\s*"
    r"(?P<literal>.+)$"
)

_PROPERTY_OPS: dict[str, Callable[[Any, Any], bool]] = {
    "==": operator.eq,
    "!=": operator.ne,
    "<": operator.lt,
    "<=": operator.le,
    ">": operator.gt,
    ">=": operator.ge,
}


def _parse_scalar(raw: str) -> object:
    """Parse one literal token: quoted string, bool, int, float, or bare word.

    Straight string/number parsing only — never ``eval``/``exec``.
    """
    raw = raw.strip()
    if len(raw) >= 2 and raw[0] == raw[-1] and raw[0] in ("'", '"'):
        return raw[1:-1]
    if raw.lower() == "true":
        return True
    if raw.lower() == "false":
        return False
    try:
        return int(raw)
    except ValueError:
        pass
    try:
        return float(raw)
    except ValueError:
        pass
    return raw


def _parse_literal(raw: str) -> object:
    """Parse a literal token, or a bracketed ``[a, b, c]`` list of them."""
    raw = raw.strip()
    if raw.startswith("[") and raw.endswith("]"):
        inner = raw[1:-1].strip()
        if not inner:
            return []
        return [_parse_scalar(item) for item in inner.split(",")]
    return _parse_scalar(raw)


def _parse_property_clause(clause: str) -> tuple[str, str, object]:
    """Tokenize one ``var OP literal`` clause. Raises :class:`PropertyParseError`."""
    match = _PROPERTY_CLAUSE_RE.match(clause.strip())
    if match is None:
        raise PropertyParseError(f"malformed property clause: {clause!r}")
    var = match.group("var")
    op = match.group("op")
    literal_raw = match.group("literal").strip()
    if op == "in" and not (literal_raw.startswith("[") and literal_raw.endswith("]")):
        raise PropertyParseError(
            f"'in' operator requires a bracketed list literal, got {literal_raw!r}"
        )
    return var, op, _parse_literal(literal_raw)


def _coerce_to_type(raw: str, target_type: type) -> object:
    """Coerce an observed string value to ``target_type``. May raise ``ValueError``."""
    if target_type is bool:
        return raw.strip().lower() in ("true", "1", "yes")
    if target_type is int:
        return int(raw)
    if target_type is float:
        return float(raw)
    return raw


def _membership(raw_value: str, items: list[object]) -> bool:
    """Check ``raw_value`` (coerced per-item) against an ``in`` list literal."""
    for item in items:
        try:
            coerced = _coerce_to_type(raw_value, type(item))
        except ValueError:
            continue
        if coerced == item:
            return True
    return False


def _compare(raw_value: str, op: str, literal: object) -> bool:
    """Evaluate one non-``in`` comparison. Never raises on a type mismatch."""
    try:
        coerced = _coerce_to_type(raw_value, type(literal))
    except ValueError:
        coerced = raw_value
    fn = _PROPERTY_OPS[op]
    try:
        return bool(fn(coerced, literal))
    except TypeError:
        return False


def _clause_holds(
    var: str, op: str, literal: object, observed_outputs: dict[str, str]
) -> bool:
    """Evaluate one already-tokenized ``var OP literal`` clause."""
    raw_value = observed_outputs[var]
    if op == "in":
        items = literal if isinstance(literal, list) else [literal]
        return _membership(raw_value, items)
    return _compare(raw_value, op, literal)


def _missing_vars(
    parsed_clauses: list[tuple[str, str, object]], observed_outputs: dict[str, str]
) -> list[str]:
    """The (sorted, deduplicated) vars a clause references but evidence lacks."""
    return sorted({var for var, _, _ in parsed_clauses if var not in observed_outputs})


class PropertyEvaluator:
    """Evaluates a PROPERTY anchor: an ``and``-joined predicate over outputs."""

    @property
    def klass(self) -> AnchorClass:
        return AnchorClass.PROPERTY

    def evaluate(self, anchor: Anchor, evidence: AnchorEvidence) -> AnchorResult:
        raw_clauses = [
            c
            for c in _PROPERTY_AND_SPLIT_RE.split(anchor.statement.strip())
            if c.strip()
        ]
        try:
            parsed_clauses = [_parse_property_clause(c) for c in raw_clauses]
        except PropertyParseError as exc:
            return _result(
                anchor,
                self.klass,
                Satisfaction.INCONCLUSIVE,
                f"malformed property statement: {exc}",
            )
        if not parsed_clauses:
            return _result(
                anchor,
                self.klass,
                Satisfaction.INCONCLUSIVE,
                f"empty property statement: {anchor.statement!r}",
            )
        missing = _missing_vars(parsed_clauses, evidence.observed_outputs)
        if missing:
            return _result(
                anchor,
                self.klass,
                Satisfaction.INCONCLUSIVE,
                f"observed_outputs missing referenced var(s): {missing}",
            )
        for var, op, literal in parsed_clauses:
            if not _clause_holds(var, op, literal, evidence.observed_outputs):
                return _result(
                    anchor,
                    self.klass,
                    Satisfaction.UNSATISFIED,
                    f"clause failed: {var} {op} {literal!r} "
                    f"(observed {evidence.observed_outputs[var]!r})",
                )
        return _result(
            anchor,
            self.klass,
            Satisfaction.SATISFIED,
            f"all {len(parsed_clauses)} clause(s) satisfied",
        )


# ---------------------------------------------------------------------------
# OracleEvaluator — observed_outputs must match differential (prior behaviour)
# ---------------------------------------------------------------------------


class OracleEvaluator:
    """Evaluates a DIFFERENTIAL_ORACLE anchor: behaviour must be unchanged vs prior."""

    @property
    def klass(self) -> AnchorClass:
        return AnchorClass.DIFFERENTIAL_ORACLE

    def evaluate(self, anchor: Anchor, evidence: AnchorEvidence) -> AnchorResult:
        if not evidence.differential:
            return _result(
                anchor,
                self.klass,
                Satisfaction.INCONCLUSIVE,
                "no prior-behaviour differential supplied",
            )
        for key, observed_value in evidence.observed_outputs.items():
            prior_value = evidence.differential.get(key)
            if prior_value != observed_value:
                return _result(
                    anchor,
                    self.klass,
                    Satisfaction.UNSATISFIED,
                    f"key {key!r} diverged from prior behaviour: "
                    f"observed={observed_value!r} prior={prior_value!r}",
                )
        return _result(
            anchor,
            self.klass,
            Satisfaction.SATISFIED,
            "observed outputs match prior differential on all observed keys",
        )


# ---------------------------------------------------------------------------
# evaluate_all — dispatch by class
# ---------------------------------------------------------------------------


@traces(spec="033", reqs=["REQ-033-C1"])
def evaluate_all(
    anchors: list[Anchor],
    evidence: AnchorEvidence,
    evaluators: Sequence[AnchorEvaluator],
) -> list[AnchorResult]:
    """Evaluate every anchor in ``anchors`` against ``evidence``, by class.

    Dispatches each anchor to the evaluator in ``evaluators`` whose
    :attr:`AnchorEvaluator.klass` matches. An anchor whose class has no
    matching evaluator is INCONCLUSIVE — it is never silently skipped,
    since a silently-skipped anchor would let :func:`~.gate.fast_path_gate`
    treat an unproven anchor as though it were not there at all.
    """
    by_class = {evaluator.klass: evaluator for evaluator in evaluators}
    results: list[AnchorResult] = []
    for anchor in anchors:
        evaluator = by_class.get(anchor.klass)
        if evaluator is None:
            results.append(
                _result(
                    anchor,
                    anchor.klass,
                    Satisfaction.INCONCLUSIVE,
                    f"no evaluator registered for anchor class {anchor.klass.value!r}",
                )
            )
            continue
        results.append(evaluator.evaluate(anchor, evidence))
    return results