scripts.factory_decision

Reference › scripts › factory_decision

The command-line interface for the SPEC-033 decision phase and the definition-of-ready (DoR) gate wiring (B8; ADR-0015 and ADR-0028). It uses the anchor store in store and the assumption ledger in ledger; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main has two subcommands. decide records a decision (_cmd_decide, printing the resulting record through _record_to_dict). construct runs a construction step for a list of requirement ids and a spec text, using a design authority (_run_construction, _cmd_construct) and producing a ConstructionArtifact. The knowledge-base directory comes from the command line or the environment (_resolve_kb_dir). 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_decision.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 decision phase + DoR gate wiring (B8, ADR-0015/0028).

Subcommands
-----------
decide      Run the decision phase over a SAVED :class:`ConstructionArtifact`
            — no LLM call. Exits 0 when ready, 1 when blocked.
construct   Invoke :class:`LLMDesignAuthority` (live LLM call) to build a
            fresh :class:`ConstructionArtifact` for a requirement set and
            save it.

Ledger/registry directory resolution (both subcommands): ``--kb`` flag >
``FACTORY_KB_DIR`` env var > ``<repo-root>/factory_kb`` — matching
``scripts/factory_assumptions.py``. Within that directory, the ledger is
``ledger.jsonl`` and the anchor registry is ``anchors.json``, unless
overridden with ``--ledger``/``--anchors``.

Examples
--------
    python scripts/factory_decision.py construct --requirements REQ-033-C5 \\
        --spec-file spec.txt --out artifact.json
    python scripts/factory_decision.py decide --requirements REQ-033-C5 \\
        --spec-file spec.txt --artifact-file artifact.json
"""

from __future__ import annotations

import argparse
import asyncio
import dataclasses
import json
import os
import sys
from collections.abc import Mapping
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parent.parent
_REPO_SRC = _REPO_ROOT / "src"
if str(_REPO_SRC) not in sys.path:
    sys.path.insert(0, str(_REPO_SRC))

from orchestrata_ultimate.factory.anchors.store import AnchorRegistry  # noqa: E402
from orchestrata_ultimate.factory.assumptions.ledger import (  # noqa: E402
    AssumptionLedger,
)
from orchestrata_ultimate.factory.decision.authority import (  # noqa: E402
    DesignAuthority,
    LLMDesignAuthority,
)
from orchestrata_ultimate.factory.decision.phase import (  # noqa: E402
    DecisionInputs,
    DetectorEngines,
    run_decision_phase,
)
from orchestrata_ultimate.factory.detectors.engine import Blacklist  # noqa: E402
from orchestrata_ultimate.factory.dor_procedure.artifact import (  # noqa: E402
    load_artifact,
    save_artifact,
)
from orchestrata_ultimate.factory.dor_procedure.models import (  # noqa: E402
    ConstructionArtifact,
)
from orchestrata_ultimate.factory.formalization.models import (  # noqa: E402
    load_formalization,
)


def _resolve_kb_dir(
    cli_value: str | None, env: Mapping[str, str] | None = None
) -> Path:
    """Resolve the factory KB directory: ``--kb`` > ``FACTORY_KB_DIR`` > repo default.

    ``env`` defaults to ``os.environ``; tests pass an explicit mapping
    instead of mutating the process environment.
    """
    if cli_value:
        return Path(cli_value)
    env = env if env is not None else os.environ
    env_value = env.get("FACTORY_KB_DIR")
    if env_value:
        return Path(env_value)
    return _REPO_ROOT / "factory_kb"


def _record_to_dict(record: object) -> dict[str, object]:
    return dataclasses.asdict(record)  # type: ignore[call-overload]


def _cmd_decide(args: argparse.Namespace) -> int:
    kb_dir = _resolve_kb_dir(args.kb)
    ledger = AssumptionLedger(
        Path(args.ledger) if args.ledger else kb_dir / "ledger.jsonl"
    )
    registry = AnchorRegistry(
        Path(args.anchors) if args.anchors else kb_dir / "anchors.json"
    )

    artifact = load_artifact(Path(args.artifact_file))
    with open(args.spec_file, encoding="utf-8") as f:
        spec_text = f.read()

    blacklist = Blacklist(Path(args.blacklist)) if args.blacklist else None
    formalization = (
        load_formalization(Path(args.formalization)) if args.formalization else None
    )
    detector_engines = DetectorEngines(
        blacklist=blacklist,
        repo_root=Path(args.repo_root) if args.repo_root else None,
        formalization=formalization,
    )

    inputs = DecisionInputs(
        requirement_ids=args.requirements.split(","),
        spec_text=spec_text,
        artifact=artifact,
        artifact_path=str(args.artifact_file),
    )
    record = run_decision_phase(
        inputs,
        ledger=ledger,
        anchor_registry=registry,
        detector_engines=detector_engines,
    )
    print(json.dumps(_record_to_dict(record), indent=2, sort_keys=True))
    return 0 if record.verdict == "ready" else 1


def _run_construction(
    authority: DesignAuthority, requirement_ids: list[str], spec_text: str
) -> ConstructionArtifact:
    """Run one B23 construction through the ADR-0028 seam.

    Typed against the ``DesignAuthority`` PROTOCOL, not the concrete
    ``LLMDesignAuthority``, which is what makes the seam real rather than
    documentary. Two things follow from that and neither did before: the type
    checker now verifies ``LLMDesignAuthority`` actually satisfies the
    Protocol at the one place it is constructed (a drift in ``construct``'s
    signature becomes a pyright error instead of a runtime surprise), and the
    substitutability the Protocol's own docstring promises -- "a test double
    that builds a ConstructionArtifact directly satisfies it just as well" --
    now has a parameter to be substituted at.
    """
    return asyncio.run(authority.construct(requirement_ids, spec_text))


def _cmd_construct(args: argparse.Namespace) -> int:
    with open(args.spec_file, encoding="utf-8") as f:
        spec_text = f.read()
    artifact = _run_construction(
        LLMDesignAuthority(volk=args.volk), args.requirements.split(","), spec_text
    )
    save_artifact(artifact, Path(args.out), overwrite=args.overwrite)
    print(f"Wrote construction artifact for {args.requirements} to {args.out}")
    return 0


def _add_decide_parser(sub: argparse._SubParsersAction) -> None:
    decide_p = sub.add_parser(
        "decide",
        help="Run the decision phase over a saved artifact (no LLM). Exits "
        "0/ready, 1/blocked.",
    )
    decide_p.add_argument("--requirements", required=True, help="Comma-separated ids.")
    decide_p.add_argument("--spec-file", required=True, help="Spec text file.")
    decide_p.add_argument(
        "--artifact-file", required=True, help="Saved ConstructionArtifact JSON."
    )
    decide_p.add_argument("--ledger", default=None, help="Ledger JSONL path override.")
    decide_p.add_argument(
        "--anchors", default=None, help="Anchor registry JSON path override."
    )
    decide_p.add_argument("--kb", default=None, help="Factory KB directory override.")
    decide_p.add_argument(
        "--formalization", default=None, help="Optional declared formalization JSON."
    )
    decide_p.add_argument(
        "--blacklist", default=None, help="Optional detector dismissal blacklist path."
    )
    decide_p.add_argument(
        "--repo-root", default=None, help="Optional repo root override for C3 rules."
    )


def _add_construct_parser(sub: argparse._SubParsersAction) -> None:
    construct_p = sub.add_parser(
        "construct",
        help="Invoke LLMDesignAuthority (live) to build a fresh artifact.",
    )
    construct_p.add_argument(
        "--requirements", required=True, help="Comma-separated ids."
    )
    construct_p.add_argument("--spec-file", required=True, help="Spec text file.")
    construct_p.add_argument("--out", required=True, help="Output artifact JSON path.")
    construct_p.add_argument(
        "--volk",
        default="design_authority",
        help="LLM profile/volk alias to route through (default: design_authority).",
    )
    construct_p.add_argument(
        "--overwrite", action="store_true", help="Allow overwriting --out."
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_decision",
        description="SPEC-033 decision phase + DoR gate wiring (B8, ADR-0015/0028).",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    _add_decide_parser(sub)
    _add_construct_parser(sub)
    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 = {
        "decide": _cmd_decide,
        "construct": _cmd_construct,
    }
    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()