orchestrata_ultimate.verification.rtm_constraints

Reference › orchestrata_ultimate › verification › rtm_constraints

Checks a requirements traceability matrix with Prolog rules (Spec 033, REQ-033-C4). It is used by scripts.generate_rtm_evidence. A caller fills an RtmFacts (requirements, requirement-to-unit trace pairs, test units, proof class per unit, minimum proof class per requirement) and calls check_rtm. The function validates every proof class against VALID_PROOF_CLASSES and raises ValueError for an unknown one. It then consults rtm_constraints.pl through the prolog adapter, asserts the facts as ground atoms (escaped with _escape), and enumerates every violation with a single all_violations query, with no cap and no truncation (ADR-0026). Each answer becomes an RtmViolation (kind, requirement, unit, needed class). Because the SWI clause database is process-global, the retract, assert and query sequence runs under a lock, and stale facts are retracted at the start of every check.

At a glance

Class diagram

Dependency diagram

API

Classes

RtmFacts (dataclass)

Decorators: dataclass(frozen=True) · Line 63

Ground facts for one RTM-constraints check.

RtmViolation (dataclass)

Decorators: dataclass(frozen=True) · Line 87

One violation counterexample — kind + the violating identifiers.

Functions

Source

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

"""RTM integrity constraints — SPEC-033 REQ-033-C4, §9 B25.

Enumerates counterexamples over the requirements-traceability-matrix facts
(never a boolean — ADR-0026, REQ-033-C4): ``check_rtm`` asserts a
:class:`RtmFacts` snapshot as GROUND facts against ``prolog/rtm_constraints.pl``
via the project's canonical janus-swi seam (``core.prolog_adapter``) and
returns every violation the rulebase's ``all_violations/1`` findall enumerates
— no cap, no truncation.

Fact shapes mirror what ``scripts/generate_rtm_evidence.py`` already computes
(requirement ids, test citations) so a caller can build an ``RtmFacts`` almost
directly from that script's ``reqs``/``cites``/(optional) ``test_results``.
"""

from __future__ import annotations

import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.prolog_adapter import PrologEngine

_RULES_PATH = Path(__file__).resolve().parent.parent / "prolog" / "rtm_constraints.pl"

# The proof-class scale, weakest to strongest — must match class_rank/2 in
# rtm_constraints.pl. Kept here too so the Python layer can validate facts
# before asserting them (a bad class name would otherwise just silently fail
# to match class_rank/2 in Prolog and mis-report as "unproven").
VALID_PROOF_CLASSES = ("failing", "cited", "tested")

# The default minimum proof class applied to any requirement that does not
# have an explicit override in RtmFacts.min_classes — ADR-0007's bar ("reached
# by a test that asserts on it"): a requirement must be traced to at least a
# CITING test, not merely a static @traces decorator claim.
DEFAULT_MIN_CLASS = "cited"

_consult_lock = threading.Lock()

# Serializes check_rtm's retract -> assert -> query critical section; the
# facts live in the process-global SWI clause DB, not per-engine.
_check_lock = threading.Lock()
_consulted = False


def _consult_rules(engine: PrologEngine) -> None:
    """Consult rtm_constraints.pl once per process (idempotent load)."""
    global _consulted
    with _consult_lock:
        if _consulted:
            return
        engine.consult(str(_RULES_PATH))
        _consulted = True


def _escape(value: str) -> str:
    """Escape a string for safe use inside a single-quoted Prolog atom."""
    return value.replace("\\", "\\\\").replace("'", "\\'")


@dataclass(frozen=True)
class RtmFacts:
    """Ground facts for one RTM-constraints check.

    Attributes:
        requirements: every requirement id known to the spec set.
        traces: ``(req_id, unit)`` pairs — a requirement traced to a test unit.
        test_units: every known/discovered test unit (guards ``dangling``;
            a trace naming a unit NOT in this set is dangling).
        proof_classes: unit -> its strongest proof class, one of
            ``VALID_PROOF_CLASSES``. A unit absent from this mapping is
            treated as having no execution evidence at all (never satisfies
            any requirement's minimum).
        min_classes: req_id -> its minimum acceptable proof class. A
            requirement absent from this mapping gets ``DEFAULT_MIN_CLASS``.
    """

    requirements: frozenset[str]
    traces: frozenset[tuple[str, str]]
    test_units: frozenset[str]
    proof_classes: dict[str, str] = field(default_factory=dict)
    min_classes: dict[str, str] = field(default_factory=dict)


@dataclass(frozen=True)
class RtmViolation:
    """One violation counterexample — kind + the violating identifiers.

    Attributes:
        kind: ``"untraced"`` | ``"unproven"`` | ``"dangling"``.
        req: the requirement id.
        unit: the test unit involved (``dangling`` only; ``None`` otherwise).
        needed_class: the minimum proof class that was not reached
            (``unproven`` only; ``None`` otherwise).
    """

    kind: str
    req: str
    unit: str | None = None
    needed_class: str | None = None


def _violation_from_dict(d: dict[str, Any]) -> RtmViolation:
    kind = d["kind"]
    return RtmViolation(
        kind=kind,
        req=d["req"],
        unit=d.get("unit"),
        needed_class=d.get("needed_class"),
    )


@traces(spec="033", reqs=["REQ-033-C4"])
def check_rtm(facts: RtmFacts) -> list[RtmViolation]:
    """Assert ``facts`` as ground Prolog facts and enumerate every violation.

    Every call retracts any RTM facts left over from a prior check on this
    engine first, so checks never leak state into each other. Enumeration is
    a single ``all_violations(Vs)`` findall — no cap, no ``once/1``, no
    truncation of the answer set (ADR-0026).

    Raises:
        ValueError: a proof class or min class in ``facts`` is not one of
            ``VALID_PROOF_CLASSES``.
    """
    for unit, cls in facts.proof_classes.items():
        if cls not in VALID_PROOF_CLASSES:
            raise ValueError(f"unknown proof class {cls!r} for unit {unit!r}")
    for req, cls in facts.min_classes.items():
        if cls not in VALID_PROOF_CLASSES:
            raise ValueError(f"unknown min class {cls!r} for requirement {req!r}")

    engine = PrologEngine()
    _consult_rules(engine)

    # The SWI clause database is process-global, so the retract -> assert ->
    # query sequence below is a critical section: two concurrent check_rtm
    # calls would wipe/corrupt each other's fact set. Serialize it.
    with _check_lock:
        engine.query_once("retract_rtm_facts")

        for req in facts.requirements:
            engine.assertz(f"requirement('{_escape(req)}')")
        for unit in facts.test_units:
            engine.assertz(f"test_unit('{_escape(unit)}')")
        for req, unit in facts.traces:
            engine.assertz(f"trace('{_escape(req)}', '{_escape(unit)}')")
        for unit, cls in facts.proof_classes.items():
            engine.assertz(f"proof_class('{_escape(unit)}', {cls})")
        for req in facts.requirements:
            cls = facts.min_classes.get(req, DEFAULT_MIN_CLASS)
            engine.assertz(f"min_class('{_escape(req)}', {cls})")

        result = engine.query_once("all_violations(Vs)")
    raw = result["Vs"] if result is not None else []
    return [_violation_from_dict(v) for v in raw]