scripts.factory_dor_procedure

Reference › scripts › factory_dor_procedure

The command-line interface for the SPEC-033 definition-of-ready (DoR) construction procedure (B23, REQ-033-C5). It uses the artifact model and the anchor export in the DoR procedure package: artifact and anchors_export. The factory narrative describes the subsystem. No module or test imports it.

Invocation: main has four subcommands. template prints a skeleton ConstructionArtifact for a requirement id and an author identity (_skeleton_artifact, _cmd_template). validate checks a filled-in artifact against the procedure (_cmd_validate). export-anchors exports the anchors from a validated artifact (_cmd_export_anchors). axes prints the axes of the procedure (_cmd_axes). With no subcommand it prints help and exits 1, and an exception prints an error and exits 1.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/factory_dor_procedure.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 DoR construction procedure (B23, REQ-033-C5).

Subcommands
-----------
template        Emit a skeleton artifact JSON for a requirement — every step
                present-but-empty, every residual axis listed with kind GAP,
                and an empty rulings checklist — so a fresh artifact
                honestly fails ``validate`` until it is actually
                constructed.
validate        Load an artifact, print every :class:`ShapeViolation`
                (including a missing/incomplete Step 6.3 rulings checklist)
                and every step-7 GAP, and exit non-zero unless
                ``procedure_ready`` (a NECESSARY, never sufficient, check —
                see ADR-0011/0015).
export-anchors  Load an artifact and write its anchor candidates (one per
                step-2 equivalence class) as JSON.
axes            Print the residual-axis catalogue with provenance tier and
                source (ADR-0028).

Examples
--------
    python scripts/factory_dor_procedure.py axes
    python scripts/factory_dor_procedure.py template --requirement REQ-033-C5 \\
        --out artifact.json
    python scripts/factory_dor_procedure.py validate artifact.json
    python scripts/factory_dor_procedure.py export-anchors artifact.json \\
        --out anchors.json
"""

from __future__ import annotations

import argparse
import dataclasses
import json
import os
import sys
from datetime import UTC, datetime
from pathlib import Path

# Ensure repo src is on path when run directly (mirrors kg_schema.py /
# run_ast_pipeline.py; also lets static analysis resolve the import without
# the editable-install .pth machinery). The repo ROOT is also required: the
# rulings checklist (factory.dor_procedure.rulings) lazily imports
# `scripts.rulings_lint` to reuse its RULINGS_DIR resolution, which only
# resolves as a package when the repo root (parent of scripts/) is on
# sys.path -- plain `python scripts/factory_dor_procedure.py ...` puts only
# the script's own directory there.
_repo_root = Path(__file__).resolve().parent.parent
_repo_src = _repo_root / "src"
if str(_repo_root) not in sys.path:
    sys.path.insert(0, str(_repo_root))
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

from orchestrata_ultimate.factory.dor_procedure.anchors_export import (  # noqa: E402
    export_anchor_candidates,
)
from orchestrata_ultimate.factory.dor_procedure.artifact import (  # noqa: E402
    load_artifact,
    save_artifact,
)
from orchestrata_ultimate.factory.dor_procedure.axes import RESIDUAL_AXES  # noqa: E402
from orchestrata_ultimate.factory.dor_procedure.models import (  # noqa: E402
    AxisAnswer,
    AxisAnswerKind,
    ConstructionArtifact,
    RulingsChecklist,
    Step1Inputs,
    Step2Classes,
    Step3Behaviors,
    Step4StateWalk,
    Step5FailureEnvelope,
    Step6Limits,
    Step7Residual,
)
from orchestrata_ultimate.factory.dor_procedure.validate import (  # noqa: E402
    procedure_ready,
)

_DEFAULT_AUTHOR = "factory-dor-procedure-cli"


def _skeleton_artifact(
    requirement_id: str, author_identity: str
) -> ConstructionArtifact:
    """Build a fresh, present-but-empty artifact for ``requirement_id``.

    Every residual axis is listed with kind GAP, and the rulings checklist
    is left empty, so ``validate`` on this output honestly fails until the
    procedure is actually carried out.
    """
    step7 = Step7Residual(
        answers=[
            AxisAnswer(axis_id=axis.axis_id, kind=AxisAnswerKind.GAP)
            for axis in RESIDUAL_AXES
        ]
    )
    return ConstructionArtifact(
        requirement_id=requirement_id,
        author_identity=author_identity,
        created_at=datetime.now(UTC).isoformat(),
        step1=Step1Inputs(),
        step2=Step2Classes(),
        step3=Step3Behaviors(),
        step4=Step4StateWalk(),
        step5=Step5FailureEnvelope(),
        step6=Step6Limits(),
        step7=step7,
        rulings_checklist=RulingsChecklist(),
    )


def _cmd_template(args: argparse.Namespace) -> int:
    author = args.author or os.environ.get("DOR_AUTHOR_IDENTITY", _DEFAULT_AUTHOR)
    artifact = _skeleton_artifact(args.requirement, author)
    save_artifact(artifact, Path(args.out), overwrite=args.overwrite)
    print(
        f"Wrote skeleton DoR construction artifact for {args.requirement} to "
        f"{args.out} (every residual axis is a GAP — validate will fail until "
        f"constructed)"
    )
    return 0


def _cmd_validate(args: argparse.Namespace) -> int:
    artifact = load_artifact(Path(args.file))
    ready, violations, gap_answers = procedure_ready(artifact)
    for violation in violations:
        print(
            f"VIOLATION step={violation.step} path={violation.path}: "
            f"{violation.statement}"
        )
    for gap in gap_answers:
        print(f"GAP axis={gap.axis_id}")
    if ready:
        print(
            "procedure_ready: True (necessary condition only, never a "
            "completeness certificate — ADR-0011/0015)"
        )
        return 0
    print(
        f"procedure_ready: False ({len(violations)} violation(s), "
        f"{len(gap_answers)} gap(s))"
    )
    return 1


def _cmd_export_anchors(args: argparse.Namespace) -> int:
    artifact = load_artifact(Path(args.file))
    candidates = export_anchor_candidates(artifact)
    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump([dataclasses.asdict(c) for c in candidates], f, indent=2)
    print(f"Wrote {len(candidates)} anchor candidate(s) to {out_path}")
    return 0


def _cmd_axes(args: argparse.Namespace) -> int:
    for axis in RESIDUAL_AXES:
        print(f"{axis.axis_id}\t[{axis.tier.value}]\t{axis.title}\t{axis.source}")
    return 0


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_dor_procedure",
        description="SPEC-033 DoR construction procedure (B23, REQ-033-C5).",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    template_p = sub.add_parser(
        "template", help="Emit a skeleton artifact JSON for a requirement."
    )
    template_p.add_argument("--requirement", required=True, help="Requirement id.")
    template_p.add_argument("--out", required=True, help="Output JSON path.")
    template_p.add_argument(
        "--author",
        default=None,
        help="Author identity (default: DOR_AUTHOR_IDENTITY env var).",
    )
    template_p.add_argument(
        "--overwrite",
        action="store_true",
        help="Allow overwriting an existing file at --out.",
    )

    validate_p = sub.add_parser(
        "validate", help="Validate an artifact's shape and print violations/gaps."
    )
    validate_p.add_argument("file", help="Artifact JSON path.")

    export_p = sub.add_parser(
        "export-anchors", help="Export anchor candidates from an artifact."
    )
    export_p.add_argument("file", help="Artifact JSON path.")
    export_p.add_argument("--out", required=True, help="Output JSON path.")

    sub.add_parser("axes", help="Print the residual-axis catalogue.")

    return parser


def main(argv: list[str] | None = None) -> None:
    """Parse ``argv``, dispatch to the matching subcommand, and ``sys.exit``."""
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {
        "template": _cmd_template,
        "validate": _cmd_validate,
        "export-anchors": _cmd_export_anchors,
        "axes": _cmd_axes,
    }
    handler = dispatch.get(args.command)
    if handler is None:
        parser.print_help()
        sys.exit(1)

    try:
        exit_code = handler(args)
    except Exception as exc:  # noqa: BLE001
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)

    sys.exit(exit_code)


if __name__ == "__main__":
    main()