orchestrata_ultimate.factory.dor_procedure.validate

Reference › orchestrata_ultimate › factory › dor_procedure › validate

validate is the shape validator for the DoR construction procedure (SPEC-033 row B23, REQ-033-C5, ADR-0028). validate_artifact is pure and deterministic: it checks that each step of a ConstructionArtifact is filled in the way the procedure requires, and returns a list of ShapeViolation records, each naming a step, a path and a statement. One private function checks each step:

• _validate_step1: unused inputs must be a subset of the inputs declared unused.
• _validate_step2: every input needs a valid class plus two invalid classes, or a valid class plus a complement class.
• _validate_step3: every class from step 2 needs a non-empty expected behaviour.
• _validate_step4: full states-by-inputs coverage plus the no-input case.
• _validate_step5: all five failure modes per dependency, or a stated reason for none.
• _validate_step6: all four limit kinds present, with not-applicable allowed when a reason is given.
• _validate_step7: exactly one answer per axis in axes, a location for SPECIFIED and a reason for NOT_APPLICABLE.
• _validate_rulings_checklist: at least one entry, or a stated reason, checked against the ids from rulings.

gaps returns the step-7 answers of kind GAP, and procedure_ready returns a tuple of readiness (shape-complete and no gaps), the violations and the gaps. The check is necessary but not sufficient (ADR-0011, ADR-0015): zero violations and zero gaps mean the shape is complete, not that the requirement is correct, and no public name claims completeness. The decision phase and the scripts.factory_dor_procedure CLI import it; tests are in tests/factory/test_dor_validate.py.

At a glance

Class diagram

Dependency diagram

API

Classes

ShapeViolation (dataclass)

Decorators: dataclass(frozen=True) · Line 33

One concrete, named shape defect.

Functions

Source

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

"""Shape validator for the SPEC-033 DoR construction procedure (B23, REQ-033-C5).

``validate_artifact`` is pure and deterministic. Every rule here checks the
*shape* the construction procedure forces (ADR-0028): a step's fields are
either filled in the way the procedure requires, or a
:class:`ShapeViolation` names exactly what is missing and where.

This is a **necessary, not sufficient**, check (ADR-0011/0015): zero
violations and zero GAPs means the artifact's shape is complete — it never
certifies that the requirement it documents is itself correct or complete.
No public name in this module may claim "complete".
"""

from __future__ import annotations

from collections import Counter
from dataclasses import dataclass

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.dor_procedure.axes import RESIDUAL_AXES
from orchestrata_ultimate.factory.dor_procedure.models import (
    AxisAnswer,
    AxisAnswerKind,
    ClassKind,
    ConstructionArtifact,
    FailureMode,
    LimitKind,
)
from orchestrata_ultimate.factory.dor_procedure.rulings import list_ruling_ids


@dataclass(frozen=True)
class ShapeViolation:
    """One concrete, named shape defect.

    ``path`` pinpoints the field (e.g. ``"step2.per_input.amount"``);
    ``statement`` says exactly what is missing or wrong — no soft language.
    """

    step: int
    path: str
    statement: str


def _na_reason_violation(
    step: int, path: str, value: str, what: str
) -> ShapeViolation | None:
    """Shared silence/N/A-reason rule used by steps 4, 5, and 6.

    ``value`` must be non-empty — silence is disallowed. If it states
    ``"N/A"``, a non-empty reason must follow the colon.
    """
    stripped = value.strip()
    if not stripped:
        return ShapeViolation(step, path, f"{what} is empty — silence is not allowed")
    if stripped.upper().startswith("N/A"):
        _, _, reason = stripped.partition(":")
        if not reason.strip():
            return ShapeViolation(
                step, path, f"{what} states N/A without a reason after the colon"
            )
    return None


def _validate_step1(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """``unused_inputs`` must be a subset of inputs declared ``used=False``."""
    violations: list[ShapeViolation] = []
    declared_unused = {i.name for i in artifact.step1.inputs if not i.used}
    for name in artifact.step1.unused_inputs:
        if name not in declared_unused:
            violations.append(
                ShapeViolation(
                    1,
                    "step1.unused_inputs",
                    f"'{name}' is listed in unused_inputs but is not declared as "
                    f"a used=False input",
                )
            )
    return violations


def _validate_step2(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """Every input needs >=1 VALID + 2 invalid classes, or VALID + COMPLEMENT."""
    violations: list[ShapeViolation] = []
    all_names = {i.name for i in artifact.step1.inputs} | set(
        artifact.step2.per_input.keys()
    )
    for name in sorted(all_names):
        path = f"step2.per_input.{name}"
        classes = artifact.step2.per_input.get(name)
        if not classes:
            violations.append(
                ShapeViolation(
                    2, path, f"no equivalence classes declared for input '{name}'"
                )
            )
            continue
        valid = sum(1 for c in classes if c.kind == ClassKind.VALID)
        below = sum(1 for c in classes if c.kind == ClassKind.INVALID_BELOW)
        above = sum(1 for c in classes if c.kind == ClassKind.INVALID_ABOVE)
        complement = sum(1 for c in classes if c.kind == ClassKind.COMPLEMENT)
        ordinal_ok = valid >= 1 and below >= 1 and above >= 1
        set_ok = valid >= 1 and complement >= 1
        if not (ordinal_ok or set_ok):
            violations.append(
                ShapeViolation(
                    2,
                    path,
                    f"input '{name}' needs >=1 VALID and (>=1 INVALID_BELOW + "
                    f">=1 INVALID_ABOVE, or >=1 COMPLEMENT) — has valid={valid} "
                    f"invalid_below={below} invalid_above={above} "
                    f"complement={complement}",
                )
            )
    return violations


def _validate_step3(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """Every class from step 2 needs a non-empty expected behaviour."""
    violations: list[ShapeViolation] = []
    for input_name, classes in artifact.step2.per_input.items():
        for eq_class in classes:
            key = f"{input_name}:{eq_class.label}"
            behavior = artifact.step3.behaviors.get(key, "")
            if not behavior.strip():
                violations.append(
                    ShapeViolation(
                        3,
                        f"step3.behaviors.{key}",
                        f"no expected behaviour for input '{input_name}' class "
                        f"'{eq_class.label}'",
                    )
                )
    return violations


def _validate_step4(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """Full states x inputs coverage, plus the no-input case.

    DELIBERATE GAP (wave-B review, deferred to B8): empty ``states``/
    ``inputs`` pass coverage vacuously — unlike step 5, there is no
    "empty requires a stated reason" rule here, because whether a state
    walk is mandatory depends on whether the requirement is stateful,
    which is a per-requirement judgement the B8 design phase owns.
    ``no_input_case`` still guards true emptiness.
    """
    violations: list[ShapeViolation] = []
    step4 = artifact.step4
    for state in step4.states:
        for inp in step4.inputs:
            key = f"{state}|{inp}"
            value = step4.transitions.get(key, "")
            violation = _na_reason_violation(
                4, f"step4.transitions.{key}", value, "transition"
            )
            if violation is not None:
                violations.append(violation)
    violation = _na_reason_violation(
        4, "step4.no_input_case", step4.no_input_case, "no_input_case"
    )
    if violation is not None:
        violations.append(violation)
    return violations


def _validate_step5(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """All 5 failure modes per dependency; empty deps needs a stated reason."""
    violations: list[ShapeViolation] = []
    step5 = artifact.step5
    if not step5.dependencies:
        if not step5.no_dependencies_reason.strip():
            violations.append(
                ShapeViolation(
                    5,
                    "step5.no_dependencies_reason",
                    "empty dependencies list requires a non-empty "
                    "no_dependencies_reason",
                )
            )
        return violations
    for dependency in step5.dependencies:
        for mode in FailureMode:
            key = f"{dependency}|{mode.value}"
            value = step5.responses.get(key, "")
            violation = _na_reason_violation(
                5, f"step5.responses.{key}", value, "failure response"
            )
            if violation is not None:
                violations.append(violation)
    return violations


def _validate_step6(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """All four limit kinds must be present (N/A-with-reason allowed)."""
    violations: list[ShapeViolation] = []
    for kind in LimitKind:
        value = artifact.step6.limits.get(kind.value, "")
        violation = _na_reason_violation(
            6, f"step6.limits.{kind.value}", value, "limit"
        )
        if violation is not None:
            violations.append(violation)
    return violations


def _validate_step7_axis_counts(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """Exactly one answer per catalogued axis — no omissions, no extras."""
    violations: list[ShapeViolation] = []
    catalogue_ids = {axis.axis_id for axis in RESIDUAL_AXES}
    counts = Counter(answer.axis_id for answer in artifact.step7.answers)
    for axis in RESIDUAL_AXES:
        count = counts.get(axis.axis_id, 0)
        if count == 0:
            violations.append(
                ShapeViolation(
                    7,
                    f"step7.answers.{axis.axis_id}",
                    f"no answer for residual axis '{axis.axis_id}'",
                )
            )
        elif count > 1:
            violations.append(
                ShapeViolation(
                    7,
                    f"step7.answers.{axis.axis_id}",
                    f"{count} answers given for residual axis '{axis.axis_id}' — "
                    f"exactly one is required",
                )
            )
    for axis_id in counts:
        if axis_id not in catalogue_ids:
            violations.append(
                ShapeViolation(
                    7,
                    f"step7.answers.{axis_id}",
                    f"'{axis_id}' is not in the residual axis catalogue",
                )
            )
    return violations


def _validate_step7_answer_fields(
    artifact: ConstructionArtifact,
) -> list[ShapeViolation]:
    """SPECIFIED needs a non-empty ``where``; NOT_APPLICABLE needs a reason."""
    violations: list[ShapeViolation] = []
    for answer in artifact.step7.answers:
        path = f"step7.answers.{answer.axis_id}"
        if answer.kind == AxisAnswerKind.SPECIFIED and not answer.where.strip():
            violations.append(
                ShapeViolation(
                    7,
                    f"{path}.where",
                    f"axis '{answer.axis_id}' is SPECIFIED but 'where' is empty",
                )
            )
        if answer.kind == AxisAnswerKind.NOT_APPLICABLE and not answer.reason.strip():
            violations.append(
                ShapeViolation(
                    7,
                    f"{path}.reason",
                    f"axis '{answer.axis_id}' is NOT_APPLICABLE but 'reason' is empty",
                )
            )
    return violations


def _validate_step7(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """Step 7's full shape check: axis coverage plus per-answer field rules."""
    return _validate_step7_axis_counts(artifact) + _validate_step7_answer_fields(
        artifact
    )


def _validate_rulings_checklist(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """The Step 6.3 rulings checklist: >=1 entry (or a stated
    ``none_apply_reason``), every entry's ``ruling_id`` a real id DERIVED
    from the rulings canon (never a hardcoded whitelist — see
    :mod:`~orchestrata_ultimate.factory.dor_procedure.rulings`), every
    entry's ``applies_because`` non-empty.

    ``step=0`` marks these violations as belonging to the checklist layered
    onto the artifact, not one of ADR-0028's seven numbered steps.
    """
    violations: list[ShapeViolation] = []
    checklist = artifact.rulings_checklist
    if not checklist.entries:
        if not checklist.none_apply_reason.strip():
            violations.append(
                ShapeViolation(
                    0,
                    "rulings_checklist.entries",
                    "rulings checklist is empty and no none_apply_reason is "
                    "given — silence is not allowed",
                )
            )
        return violations
    known_ids = set(list_ruling_ids())
    for i, entry in enumerate(checklist.entries):
        path = f"rulings_checklist.entries[{i}]"
        if known_ids and entry.ruling_id not in known_ids:
            violations.append(
                ShapeViolation(
                    0,
                    f"{path}.ruling_id",
                    f"'{entry.ruling_id}' is not a known ruling id in the "
                    f"rulings canon",
                )
            )
        if not entry.applies_because.strip():
            violations.append(
                ShapeViolation(
                    0,
                    f"{path}.applies_because",
                    f"entry for '{entry.ruling_id}' has an empty applies_because",
                )
            )
    return violations


@traces(spec="033", reqs=["REQ-033-C5"])
def validate_artifact(artifact: ConstructionArtifact) -> list[ShapeViolation]:
    """Validate the machine-checkable shape of ``artifact``.

    Pure, deterministic. An empty return means the artifact's shape is
    complete across all seven numbered steps plus the Step 6.3 rulings
    checklist — it never certifies that the underlying requirement is
    itself correct or complete (ADR-0011/0015).
    """
    violations: list[ShapeViolation] = []
    violations.extend(_validate_step1(artifact))
    violations.extend(_validate_step2(artifact))
    violations.extend(_validate_step3(artifact))
    violations.extend(_validate_step4(artifact))
    violations.extend(_validate_step5(artifact))
    violations.extend(_validate_step6(artifact))
    violations.extend(_validate_step7(artifact))
    violations.extend(_validate_rulings_checklist(artifact))
    return violations


def gaps(artifact: ConstructionArtifact) -> list[AxisAnswer]:
    """The step-7 answers with kind ``GAP`` in ``artifact``."""
    return [a for a in artifact.step7.answers if a.kind == AxisAnswerKind.GAP]


@traces(spec="033", reqs=["REQ-033-C5"])
def procedure_ready(
    artifact: ConstructionArtifact,
) -> tuple[bool, list[ShapeViolation], list[AxisAnswer]]:
    """Whether ``artifact`` is ready: shape-complete AND zero GAPs.

    This is a **necessary condition only** (ADR-0011/0015). It can *refute*
    readiness — a violation or a GAP proves the procedure was not carried
    through to completion — but it can never *certify* completeness: the
    absence of a violation means this construction procedure found nothing
    it could name as missing, not that the requirement was fully thought
    through.
    """
    violations = validate_artifact(artifact)
    gap_answers = gaps(artifact)
    ready = not violations and not gap_answers
    return ready, violations, gap_answers