orchestrata_ultimate.factory.dor_procedure.artifact

Reference › orchestrata_ultimate › factory › dor_procedure › artifact

artifact persists the ConstructionArtifact of the DoR construction procedure (SPEC-033 row B23, REQ-033-C5). The artifact is retained evidence under ADR-0028, so save_artifact writes JSON atomically (temporary file and os.replace) and refuses to overwrite an existing path unless overwrite=True is passed. load_artifact reads it back with an exact round trip. The private _artifact_to_dict and the _step1_from_dict to _step7_from_dict, _rulings_checklist_from_dict and _artifact_from_dict helpers rebuild the nested frozen dataclasses from models; a missing rulings checklist defaults to empty. The module supports schema version 1. The design authority and the scripts.factory_decision and scripts.factory_dor_procedure CLIs import it; tests are in tests/factory/test_dor_artifact.py.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Persistence for :class:`ConstructionArtifact` (B23, REQ-033-C5).

JSON, atomic write (temp file + ``os.replace``, mirroring the pattern in
:mod:`orchestrata_ultimate.factory.assumptions.ledger`), exact round-trip.
The artifact is retained EVIDENCE (ADR-0028) — :func:`save_artifact` refuses
to overwrite an existing path unless ``overwrite=True`` is passed explicitly.
"""

from __future__ import annotations

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

from orchestrata_ultimate.factory.dor_procedure.models import (
    AxisAnswer,
    AxisAnswerKind,
    ClassKind,
    ConstructionArtifact,
    EquivalenceClass,
    InputSpec,
    RulingChecklistEntry,
    RulingsChecklist,
    Step1Inputs,
    Step2Classes,
    Step3Behaviors,
    Step4StateWalk,
    Step5FailureEnvelope,
    Step6Limits,
    Step7Residual,
)

_SUPPORTED_SCHEMA_VERSION = 1


def _artifact_to_dict(artifact: ConstructionArtifact) -> dict[str, Any]:
    return dataclasses.asdict(artifact)


def _step1_from_dict(step1_data: dict[str, Any]) -> Step1Inputs:
    return Step1Inputs(
        inputs=[InputSpec(**i) for i in step1_data["inputs"]],
        unused_inputs=list(step1_data["unused_inputs"]),
    )


def _step2_from_dict(step2_data: dict[str, Any]) -> Step2Classes:
    return Step2Classes(
        per_input={
            name: [
                EquivalenceClass(
                    label=c["label"],
                    kind=ClassKind(c["kind"]),
                    description=c["description"],
                    boundary=c["boundary"],
                )
                for c in classes
            ]
            for name, classes in step2_data["per_input"].items()
        }
    )


def _step4_from_dict(step4_data: dict[str, Any]) -> Step4StateWalk:
    return Step4StateWalk(
        states=list(step4_data["states"]),
        inputs=list(step4_data["inputs"]),
        transitions=dict(step4_data["transitions"]),
        no_input_case=step4_data["no_input_case"],
    )


def _step5_from_dict(step5_data: dict[str, Any]) -> Step5FailureEnvelope:
    return Step5FailureEnvelope(
        dependencies=list(step5_data["dependencies"]),
        responses=dict(step5_data["responses"]),
        no_dependencies_reason=step5_data["no_dependencies_reason"],
    )


def _step7_from_dict(step7_data: dict[str, Any]) -> Step7Residual:
    return Step7Residual(
        answers=[
            AxisAnswer(
                axis_id=a["axis_id"],
                kind=AxisAnswerKind(a["kind"]),
                where=a["where"],
                reason=a["reason"],
            )
            for a in step7_data["answers"]
        ]
    )


def _rulings_checklist_from_dict(
    data: dict[str, Any] | None,
) -> RulingsChecklist:
    """Build the rulings checklist, defaulting to empty when absent.

    A pre-Step-6.3 artifact JSON has no ``rulings_checklist`` key at all —
    loading it must not crash; it must load as an artifact whose checklist
    is empty, which ``validate_artifact`` then names as incomplete (the
    enforcement point, not the loader).
    """
    if data is None:
        return RulingsChecklist()
    return RulingsChecklist(
        entries=[
            RulingChecklistEntry(
                ruling_id=e["ruling_id"], applies_because=e["applies_because"]
            )
            for e in data.get("entries", [])
        ],
        none_apply_reason=data.get("none_apply_reason", ""),
    )


def _artifact_from_dict(data: dict[str, Any]) -> ConstructionArtifact:
    version = data.get("schema_version")
    if version != _SUPPORTED_SCHEMA_VERSION:
        raise ValueError(
            f"Unsupported ConstructionArtifact schema_version: {version!r} "
            f"(this loader understands {_SUPPORTED_SCHEMA_VERSION!r})"
        )

    return ConstructionArtifact(
        requirement_id=data["requirement_id"],
        author_identity=data["author_identity"],
        created_at=data["created_at"],
        step1=_step1_from_dict(data["step1"]),
        step2=_step2_from_dict(data["step2"]),
        step3=Step3Behaviors(behaviors=dict(data["step3"]["behaviors"])),
        step4=_step4_from_dict(data["step4"]),
        step5=_step5_from_dict(data["step5"]),
        step6=Step6Limits(limits=dict(data["step6"]["limits"])),
        step7=_step7_from_dict(data["step7"]),
        rulings_checklist=_rulings_checklist_from_dict(data.get("rulings_checklist")),
        schema_version=data["schema_version"],
    )


def save_artifact(
    artifact: ConstructionArtifact, path: Path | str, *, overwrite: bool = False
) -> None:
    """Save ``artifact`` as JSON at ``path``, atomically.

    Refuses to overwrite an existing file unless ``overwrite=True`` is
    passed explicitly — the artifact is retained EVIDENCE (ADR-0028).

    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 DoR construction artifact at "
            f"{target} (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(_artifact_to_dict(artifact), f, indent=2, sort_keys=True)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, target)
    finally:
        if tmp.exists():
            tmp.unlink()


def load_artifact(path: Path | str) -> ConstructionArtifact:
    """Load a :class:`ConstructionArtifact` 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 _artifact_from_dict(data)