orchestrata_ultimate.factory.contracts.models

Reference › orchestrata_ultimate › factory › contracts › models

models defines the smallest design-node contract schema that can say which modules and symbols a node owns and which requirements it implements (SPEC-033 row B3, ADR-0007). DesignNode holds a node id, title, modules, symbols and requirements; DesignContract holds a contract id, a change reference, a tuple of nodes and a schema_version (currently 1). Persistence is JSON with an atomic write (temporary file and os.replace): save_contract writes and load_contract reads and validates. validate_contract raises ContractValidationError for non-unique node ids, a module claimed by more than one node, or a malformed requirement id (_is_valid_req_id). Validation runs on load, not on construction, so a contract built in memory may be inconsistent until written. The conformance engine, trace_validation and the scripts.factory_conformance and scripts.factory_trace_check CLIs use it.

At a glance

Class diagram

Dependency diagram

API

Classes

DesignNode (dataclass)

Decorators: dataclass(frozen=True) · Line 33

One design node's contract.

DesignContract (dataclass)

Decorators: dataclass(frozen=True) · Line 53

A change's full set of design nodes.

ContractValidationError

Bases: ValueError · Line 67

Raised when a :class:DesignContract fails structural validation.

Functions

Source

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

"""The minimal design-node contract schema (SPEC-033 B3, ADR-0007).

ADR-0007 requires checking that a traced unit's AST "lies within the cited
design node's contract" -- that needs a contract representation to check
against, which is what this module defines: the smallest schema that can
name which modules/symbols a design node owns and which requirements it
implements, nothing more (orchestrator-settled; see
``tasks/b3-trace-validation/instructions.md``).

Persistence mirrors the pattern in
:mod:`orchestrata_ultimate.factory.dor_procedure.artifact` -- JSON, atomic
write (temp file + ``os.replace``), exact round-trip. Structural validation
(unique node ids, no module double-claimed, well-formed requirement ids) runs
on :func:`load_contract`, not on construction -- a ``DesignContract`` built
in-process (e.g. by ``scripts/factory_trace_check.py template``) is not
required to already be internally consistent before it is written out.
"""

from __future__ import annotations

import json
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from orchestrata_ultimate.factory.assumptions.sources import _REQ_HEADING_RE

_SUPPORTED_SCHEMA_VERSION = 1


@dataclass(frozen=True)
class DesignNode:
    """One design node's contract.

    ``node_id`` is a kebab-case identifier, unique within a
    :class:`DesignContract` (enforced on load). ``modules`` are repo-relative
    ``.py`` paths this node owns -- a module belongs to AT MOST one node
    across the whole contract (also enforced on load). ``symbols`` are
    optional qualified names within those modules; an empty tuple means "the
    whole module" is in-contract. ``requirements`` are the ``REQ-...`` ids
    this node implements.
    """

    node_id: str
    title: str
    modules: tuple[str, ...] = ()
    symbols: tuple[str, ...] = ()
    requirements: tuple[str, ...] = ()


@dataclass(frozen=True)
class DesignContract:
    """A change's full set of design nodes.

    ``change_ref`` is the branch/commit context the contract was authored
    against (free text, e.g. a branch name or commit sha) -- not itself
    structurally validated.
    """

    contract_id: str
    change_ref: str
    nodes: tuple[DesignNode, ...] = ()
    schema_version: int = 1


class ContractValidationError(ValueError):
    """Raised when a :class:`DesignContract` fails structural validation."""


def _is_valid_req_id(req_id: str) -> bool:
    """True iff ``req_id`` matches the repo's ``REQ-...`` id convention.

    Reuses -- imports, does not copy -- the ``ref`` capture group of
    :data:`orchestrata_ultimate.factory.assumptions.sources._REQ_HEADING_RE`,
    which is built to scan ``### REQ-...`` markdown headings. A bare id is
    checked by matching a synthetic one-line heading built from it: the
    match must consume the *entire* id as the ``ref`` group with no trailing
    ``text`` group left over, so ``req_id`` has to be exactly a valid ref,
    not merely a prefix of one.
    """
    match = _REQ_HEADING_RE.match(f"# {req_id}")
    return bool(match and match.group("ref") == req_id and not match.group("text"))


def validate_contract(contract: DesignContract) -> None:
    """Raise :class:`ContractValidationError` if ``contract`` is invalid.

    Exactly the three structural checks the brief specifies -- no more:

    - node ids are unique within the contract;
    - no module is claimed by more than one node;
    - every requirement id matches the repo's ``REQ-...`` id convention.
    """
    seen_ids: set[str] = set()
    module_owner: dict[str, str] = {}
    for node in contract.nodes:
        if node.node_id in seen_ids:
            raise ContractValidationError(
                f"duplicate node_id in contract: {node.node_id!r}"
            )
        seen_ids.add(node.node_id)
        for module in node.modules:
            if module in module_owner:
                raise ContractValidationError(
                    f"module {module!r} claimed by both node "
                    f"{module_owner[module]!r} and node {node.node_id!r}"
                )
            module_owner[module] = node.node_id
        for req_id in node.requirements:
            if not _is_valid_req_id(req_id):
                raise ContractValidationError(
                    f"node {node.node_id!r} has a malformed requirement id: "
                    f"{req_id!r}"
                )


def _node_to_dict(node: DesignNode) -> dict[str, Any]:
    return {
        "node_id": node.node_id,
        "title": node.title,
        "modules": list(node.modules),
        "symbols": list(node.symbols),
        "requirements": list(node.requirements),
    }


def _node_from_dict(data: dict[str, Any]) -> DesignNode:
    return DesignNode(
        node_id=data["node_id"],
        title=data["title"],
        modules=tuple(data.get("modules", ())),
        symbols=tuple(data.get("symbols", ())),
        requirements=tuple(data.get("requirements", ())),
    )


def _contract_to_dict(contract: DesignContract) -> dict[str, Any]:
    return {
        "contract_id": contract.contract_id,
        "change_ref": contract.change_ref,
        "nodes": [_node_to_dict(n) for n in contract.nodes],
        "schema_version": contract.schema_version,
    }


def _contract_from_dict(data: dict[str, Any]) -> DesignContract:
    version = data.get("schema_version")
    if version != _SUPPORTED_SCHEMA_VERSION:
        raise ValueError(
            f"Unsupported DesignContract schema_version: {version!r} "
            f"(this loader understands {_SUPPORTED_SCHEMA_VERSION!r})"
        )
    return DesignContract(
        contract_id=data["contract_id"],
        change_ref=data["change_ref"],
        nodes=tuple(_node_from_dict(n) for n in data.get("nodes", ())),
        schema_version=version,
    )


def save_contract(contract: DesignContract, path: Path | str) -> None:
    """Save ``contract`` as JSON at ``path``, atomically (temp + ``os.replace``)."""
    target = Path(path)
    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(_contract_to_dict(contract), f, indent=2, sort_keys=True)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, target)
    finally:
        if tmp.exists():
            tmp.unlink()


def load_contract(path: Path | str) -> DesignContract:
    """Load + structurally validate a :class:`DesignContract` from JSON at ``path``.

    Raises:
        ValueError: If the stored ``schema_version`` is not one this loader
            understands.
        ContractValidationError: If the loaded contract fails the structural
            checks in :func:`validate_contract`.
    """
    with open(Path(path), encoding="utf-8") as f:
        data = json.load(f)
    contract = _contract_from_dict(data)
    validate_contract(contract)
    return contract


__all__ = [
    "ContractValidationError",
    "DesignContract",
    "DesignNode",
    "load_contract",
    "save_contract",
    "validate_contract",
]