orchestrata_ultimate.factory.formalization.models

Reference › orchestrata_ultimate › factory › formalization › models

models defines the declared-formalization schema for C4 (SPEC-033 row B26, REQ-033-C4, ADR-0022). A Formalization is a finite constraint-satisfaction problem written against a requirement's acceptance criteria. Its parts are Variable records (a lowercase name, a Domain of unique string values and an observable flag) and Constraint records of three ConstraintKind shapes: REQUIRE and FORBID (allowed or forbidden tuples over variables) and IMPLIES (an antecedent pair implying values of a consequent variable). Each constraint carries a source naming the sentence it formalises, so a finding can show a human which sentences conflict.

Validation runs loudly in the __post_init__ methods: constraint variables must exist, tuple arities must match, every literal must belong to its variable's domain, and at least one variable must be observable, because otherwise under-constraint detection would be vacuous. All dataclasses are frozen. formalization_to_dict and formalization_from_dict convert to and from JSON-ready dictionaries, and save_formalization (atomic, via a temporary file and os.replace) and load_formalization persist them at schema version 1. c4, the decision phase and the scripts.factory_c4 and scripts.factory_decision CLIs import it.

At a glance

Class diagram

Dependency diagram

API

Classes

Domain (dataclass)

Decorators: dataclass(frozen=True) · Line 42

A finite, non-empty domain of unique string values for a Variable.

Variable (dataclass)

Decorators: dataclass(frozen=True) · Line 55

A named, typed variable in a declared formalization.

ConstraintKind (enum)

Bases: StrEnum · Line 75

The three constraint shapes a declared formalization can express.

Constraint (dataclass)

Decorators: dataclass(frozen=True) · Line 84

A single constraint over one or more variables.

Formalization (dataclass)

Decorators: dataclass(frozen=True) · Line 228

A requirement's declared formalization: variables + constraints.

Functions

Source

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

"""Declared-formalization schema for SPEC-033 C4 (B26, REQ-033-C4, ADR-0022).

A :class:`Formalization` is a finite constraint-satisfaction problem: named,
typed ``Variable``s over finite ``Domain``s, and ``Constraint``s that pin down
allowed/forbidden assignments or conditional implications. It is authored
against a requirement's ACs -- each :class:`Constraint` carries a ``source``
naming the sentence it formalizes, since C4's findings must let a human see
WHICH sentences conflict (spec §6).

Structural validation runs in :meth:`Formalization.__post_init__` (never
silently, per the "loud, not silent" discipline in the B26 brief):
constraint variables must exist, tuple arities must match the constraint's
``vars``, every literal must be a member of its variable's domain, and at
least one variable must be ``observable`` -- otherwise under-constraint
detection would be undecidable by vacuity, which must raise rather than
silently report "fully pinned".

All dataclasses are frozen (immutable records), matching the convention in
:mod:`orchestrata_ultimate.factory.assumptions.models` and
:mod:`orchestrata_ultimate.factory.dor_procedure.models`. Persistence
(:func:`save_formalization` / :func:`load_formalization`) is JSON, atomic
write (temp file + ``os.replace``), exact round-trip, mirroring
:mod:`orchestrata_ultimate.factory.dor_procedure.artifact`.
"""

from __future__ import annotations

import json
import os
import re
from dataclasses import dataclass
from enum import StrEnum
from pathlib import Path
from typing import Any

_SUPPORTED_SCHEMA_VERSION = 1

_VARIABLE_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*$")


@dataclass(frozen=True)
class Domain:
    """A finite, non-empty domain of unique string values for a Variable."""

    values: tuple[str, ...]

    def __post_init__(self) -> None:
        if not self.values:
            raise ValueError("Domain.values must be non-empty")
        if len(set(self.values)) != len(self.values):
            raise ValueError(f"Domain.values must be unique, got {self.values!r}")


@dataclass(frozen=True)
class Variable:
    """A named, typed variable in a declared formalization.

    ``observable`` marks whether this variable's value is AC-observable --
    materiality (ADR-0022's under-constraint criterion) is defined over
    observable variables only.
    """

    name: str
    domain: Domain
    observable: bool

    def __post_init__(self) -> None:
        if not _VARIABLE_NAME_RE.match(self.name):
            raise ValueError(
                f"Variable.name must be lower_snake (e.g. 'page_size'), "
                f"got {self.name!r}"
            )


class ConstraintKind(StrEnum):
    """The three constraint shapes a declared formalization can express."""

    REQUIRE = "require"
    FORBID = "forbid"
    IMPLIES = "implies"


@dataclass(frozen=True)
class Constraint:
    """A single constraint over one or more variables.

    ``REQUIRE``/``FORBID`` constraints use ``vars`` + ``tuples`` (the allowed
    or forbidden assignment tuples, in ``vars`` order). ``IMPLIES``
    constraints use ``antecedent`` (a single ``(var, value)`` pair) +
    ``consequent_var`` + ``consequent_values`` (the set of values the
    consequent variable is restricted to when the antecedent holds).

    ``source`` is mandatory provenance -- which AC/sentence this constraint
    formalizes -- so a contradiction finding's core can name the actual
    conflicting sentences, not just opaque ids.
    """

    constraint_id: str
    kind: ConstraintKind
    source: str
    vars: tuple[str, ...] = ()
    tuples: tuple[tuple[str, ...], ...] = ()
    antecedent: tuple[str, str] | None = None
    consequent_var: str | None = None
    consequent_values: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        if not self.constraint_id:
            raise ValueError("Constraint.constraint_id must not be empty")
        if not self.source:
            raise ValueError(
                f"Constraint {self.constraint_id!r}: source (provenance) "
                f"must not be empty"
            )
        if self.kind in (ConstraintKind.REQUIRE, ConstraintKind.FORBID):
            if not self.vars:
                raise ValueError(
                    f"Constraint {self.constraint_id!r} ({self.kind.value}): "
                    f"vars must not be empty"
                )
            if not self.tuples:
                raise ValueError(
                    f"Constraint {self.constraint_id!r} ({self.kind.value}): "
                    f"tuples must not be empty"
                )
        else:  # IMPLIES
            if self.antecedent is None:
                raise ValueError(
                    f"Constraint {self.constraint_id!r} (implies): "
                    f"antecedent is required"
                )
            if self.consequent_var is None:
                raise ValueError(
                    f"Constraint {self.constraint_id!r} (implies): "
                    f"consequent_var is required"
                )
            if not self.consequent_values:
                raise ValueError(
                    f"Constraint {self.constraint_id!r} (implies): "
                    f"consequent_values must not be empty"
                )


def _validate_require_forbid(
    constraint: Constraint, variables: dict[str, Variable]
) -> None:
    for var_name in constraint.vars:
        if var_name not in variables:
            raise ValueError(
                f"Constraint {constraint.constraint_id!r} references unknown "
                f"variable {var_name!r}"
            )
    arity = len(constraint.vars)
    for tup in constraint.tuples:
        if len(tup) != arity:
            raise ValueError(
                f"Constraint {constraint.constraint_id!r}: tuple {tup!r} has arity "
                f"{len(tup)}, expected {arity} (len(vars))"
            )
        for var_name, literal in zip(constraint.vars, tup, strict=True):
            domain = variables[var_name].domain
            if literal not in domain.values:
                raise ValueError(
                    f"Constraint {constraint.constraint_id!r}: value {literal!r} for "
                    f"variable {var_name!r} is not in its domain {domain.values!r}"
                )


def _validate_implies(constraint: Constraint, variables: dict[str, Variable]) -> None:
    assert constraint.antecedent is not None
    assert constraint.consequent_var is not None
    ant_var, ant_value = constraint.antecedent
    if ant_var not in variables:
        raise ValueError(
            f"Constraint {constraint.constraint_id!r}: antecedent references "
            f"unknown variable {ant_var!r}"
        )
    if ant_value not in variables[ant_var].domain.values:
        raise ValueError(
            f"Constraint {constraint.constraint_id!r}: antecedent value {ant_value!r} "
            f"is not in {ant_var!r}'s domain {variables[ant_var].domain.values!r}"
        )
    if constraint.consequent_var not in variables:
        raise ValueError(
            f"Constraint {constraint.constraint_id!r}: consequent_var "
            f"{constraint.consequent_var!r} is unknown"
        )
    consequent_domain = variables[constraint.consequent_var].domain
    for literal in constraint.consequent_values:
        if literal not in consequent_domain.values:
            raise ValueError(
                f"Constraint {constraint.constraint_id!r}: consequent value "
                f"{literal!r} is not in {constraint.consequent_var!r}'s "
                f"domain {consequent_domain.values!r}"
            )


def _validate_formalization(
    variables: tuple[Variable, ...], constraints: tuple[Constraint, ...]
) -> None:
    seen_names: set[str] = set()
    for variable in variables:
        if variable.name in seen_names:
            raise ValueError(f"duplicate Variable.name {variable.name!r}")
        seen_names.add(variable.name)
    by_name = {v.name: v for v in variables}

    seen_ids: set[str] = set()
    for constraint in constraints:
        if constraint.constraint_id in seen_ids:
            raise ValueError(
                f"duplicate Constraint.constraint_id {constraint.constraint_id!r}"
            )
        seen_ids.add(constraint.constraint_id)
        if constraint.kind in (ConstraintKind.REQUIRE, ConstraintKind.FORBID):
            _validate_require_forbid(constraint, by_name)
        else:
            _validate_implies(constraint, by_name)

    if not any(v.observable for v in variables):
        raise ValueError(
            "Formalization must declare at least one observable variable -- "
            "otherwise under-constraint detection is undecidable by vacuity"
        )


@dataclass(frozen=True)
class Formalization:
    """A requirement's declared formalization: variables + constraints.

    Structural validation runs at construction time (``__post_init__``):
    every constraint's variables must exist, tuple arities must match,
    every literal must be a domain member, and at least one variable must
    be observable.
    """

    requirement_id: str
    variables: tuple[Variable, ...]
    constraints: tuple[Constraint, ...]
    schema_version: int = _SUPPORTED_SCHEMA_VERSION

    def __post_init__(self) -> None:
        if not self.requirement_id:
            raise ValueError("Formalization.requirement_id must not be empty")
        _validate_formalization(self.variables, self.constraints)


def _domain_to_dict(domain: Domain) -> dict[str, Any]:
    return {"values": list(domain.values)}


def _variable_to_dict(variable: Variable) -> dict[str, Any]:
    return {
        "name": variable.name,
        "domain": _domain_to_dict(variable.domain),
        "observable": variable.observable,
    }


def _constraint_to_dict(constraint: Constraint) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "constraint_id": constraint.constraint_id,
        "kind": constraint.kind.value,
        "source": constraint.source,
    }
    if constraint.kind in (ConstraintKind.REQUIRE, ConstraintKind.FORBID):
        payload["vars"] = list(constraint.vars)
        payload["tuples"] = [list(t) for t in constraint.tuples]
    else:
        assert constraint.antecedent is not None
        payload["antecedent"] = list(constraint.antecedent)
        payload["consequent_var"] = constraint.consequent_var
        payload["consequent_values"] = list(constraint.consequent_values)
    return payload


def formalization_to_dict(formalization: Formalization) -> dict[str, Any]:
    """Convert ``formalization`` to a JSON-serializable dict."""
    return {
        "requirement_id": formalization.requirement_id,
        "schema_version": formalization.schema_version,
        "variables": [_variable_to_dict(v) for v in formalization.variables],
        "constraints": [_constraint_to_dict(c) for c in formalization.constraints],
    }


def _variable_from_dict(data: dict[str, Any]) -> Variable:
    return Variable(
        name=data["name"],
        domain=Domain(values=tuple(data["domain"]["values"])),
        observable=data["observable"],
    )


def _constraint_from_dict(data: dict[str, Any]) -> Constraint:
    kind = ConstraintKind(data["kind"])
    if kind in (ConstraintKind.REQUIRE, ConstraintKind.FORBID):
        return Constraint(
            constraint_id=data["constraint_id"],
            kind=kind,
            source=data["source"],
            vars=tuple(data["vars"]),
            tuples=tuple(tuple(t) for t in data["tuples"]),
        )
    antecedent = data["antecedent"]
    return Constraint(
        constraint_id=data["constraint_id"],
        kind=kind,
        source=data["source"],
        antecedent=(antecedent[0], antecedent[1]),
        consequent_var=data["consequent_var"],
        consequent_values=tuple(data["consequent_values"]),
    )


def formalization_from_dict(data: dict[str, Any]) -> Formalization:
    """Reconstruct a :class:`Formalization` from :func:`formalization_to_dict` output.

    Raises:
        ValueError: If ``schema_version`` is not one this loader understands.
    """
    version = data.get("schema_version")
    if version != _SUPPORTED_SCHEMA_VERSION:
        raise ValueError(
            f"Unsupported Formalization schema_version: {version!r} "
            f"(this loader understands {_SUPPORTED_SCHEMA_VERSION!r})"
        )
    return Formalization(
        requirement_id=data["requirement_id"],
        variables=tuple(_variable_from_dict(v) for v in data["variables"]),
        constraints=tuple(_constraint_from_dict(c) for c in data["constraints"]),
        schema_version=data["schema_version"],
    )


def save_formalization(
    formalization: Formalization, path: Path | str, *, overwrite: bool = True
) -> None:
    """Save ``formalization`` as JSON at ``path``, atomically (temp + ``os.replace``).

    Raises:
        FileExistsError: If ``path`` already exists and ``overwrite`` is False.
    """
    target = Path(path)
    if target.exists() and not overwrite:
        raise FileExistsError(
            f"refusing to overwrite existing formalization at {target} "
            f"(pass overwrite=True to replace it)"
        )
    target.parent.mkdir(parents=True, exist_ok=True)
    tmp = target.with_name(f"{target.name}.tmp.{os.getpid()}")
    try:
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(formalization_to_dict(formalization), f, indent=2, sort_keys=True)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, target)
    finally:
        if tmp.exists():
            tmp.unlink()


def load_formalization(path: Path | str) -> Formalization:
    """Load a :class:`Formalization` from JSON at ``path``.

    Raises:
        ValueError: If the stored ``schema_version`` is not one this loader
            understands.
    """
    with open(Path(path), encoding="utf-8") as f:
        data = json.load(f)
    return formalization_from_dict(data)