orchestrata_ultimate.factory.decision.phase

Reference › orchestrata_ultimate › factory › decision › phase

phase is the decision phase of the factory (SPEC-033 row B8, ADR-0015 as amended by ADR-0028). It wires the assumption ledger (B13), anchors (B15), the C3 and C4 detectors (B22, B26) and the DoR construction procedure (B23) into one Definition-of-Ready verdict. run_decision_phase is pure orchestration: it is deterministic given its inputs and the ledger and registry it reads, and it makes no LLM call; the LLM boundary is authority, which produces the ConstructionArtifact this module consumes.

Inputs are bundled in DecisionInputs (requirement ids, spec text, the artifact) and DetectorEngines (rules, disabled rules, blacklist, repository root and an optional declared formalization). Each blocking source reports BlockingItem records (source, id, statement) in a fixed order, and a shape-invalid artifact stops later stages: _shape_blocking (artifact shape from dor_procedure.validate), _c4_blocking and _c3_blocking (detectors, tier 1 only), _assumption_blocking (any assumption for these requirements that is not PINNED), _residual_blocking (an unresolved GAP answer from step 7) and _anchor_blocking (a requirement with no RATIFIED anchor blocks, and its candidate anchors are queued). The result is a DecisionRecord with a verdict of ready or blocked, the blocking items, a ratification queue of RatificationItem records (anchor proposals and assumption escalations), provenance and a summary. The scripts.factory_decision CLI imports it; tests are in tests/factory/test_decision_phase.py.

At a glance

Class diagram

Dependency diagram

API

Classes

DecisionInputs (dataclass)

Decorators: dataclass(frozen=True) · Line 68

Everything :func:run_decision_phase needs about one decision run.

DetectorEngines (dataclass)

Decorators: dataclass(frozen=True) · Line 83

Bundle of C3/C4 detector configuration for :func:run_decision_phase.

BlockingItem (dataclass)

Decorators: dataclass(frozen=True) · Line 105

One named, human-readable blocking defect.

RatificationItem (dataclass)

Decorators: dataclass(frozen=True) · Line 120

One item queued for human ratification.

DecisionRecord (dataclass)

Decorators: dataclass(frozen=True) · Line 136

The result of one :func:run_decision_phase call.

Functions

Source

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

"""The SPEC-033 decision phase (B8): wires B13/B15/B22/B23/B26 into one
DoR verdict, per ADR-0015 (amended by ADR-0028).

:func:`run_decision_phase` is pure orchestration — deterministic given its
inputs and the state of the ledger/registry it reads, and it makes **no LLM
call** anywhere in this module (the LLM boundary is
:mod:`~orchestrata_ultimate.factory.decision.authority`, which produces the
:class:`~orchestrata_ultimate.factory.dor_procedure.models.
ConstructionArtifact` this module consumes).

Every blocking source in ADR-0015/0028's gate reports through the same
:class:`BlockingItem` contract, run in this fixed order (later stages never
run once an earlier one already blocks the artifact's shape — a
shape-invalid artifact cannot be scored by anything downstream):

1. **shape** — B23's :func:`~...dor_procedure.validate.validate_artifact`
   (any :class:`~...dor_procedure.validate.ShapeViolation` blocks alone).
2. **c4** then **c3** — the model-free detectors (B26 then B22, "runs ahead
   of C2" per spec §6 / ADR-0022).
3. **assumption** — B13's ledger: any assumption for these requirements that
   is not yet ``PINNED`` blocks (mirrors
   :func:`~orchestrata_ultimate.factory.assumptions.dor.check_dor`'s own
   "ready iff every assumption is PINNED" rule, scoped to these requirements).
4. **residual** — B23 step-7 answers of kind ``GAP`` (Example Mapping's red
   card: a recorded unresolved question is never cheaper to omit).
5. **anchor** — ADR-0015's positive human-ratified floor: a requirement with
   no RATIFIED anchor blocks, and its candidate anchors (from
   :mod:`~...dor_procedure.anchors_export`) are queued for ratification.

# B14/C2 seam: interpretation-mutation findings plug in here as a further
# detector stage once built — same BlockingItem contract.
"""

from __future__ import annotations

import time
from collections.abc import Set as AbstractSet
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.anchors.models import AnchorStatus
from orchestrata_ultimate.factory.anchors.store import AnchorRegistry
from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger
from orchestrata_ultimate.factory.assumptions.models import AssumptionStatus
from orchestrata_ultimate.factory.detectors.engine import (
    Blacklist,
    run_detectors,
    tier1_gate,
)
from orchestrata_ultimate.factory.detectors.models import RuleMeta
from orchestrata_ultimate.factory.detectors.rules import ALL_RULES, RuleFn
from orchestrata_ultimate.factory.dor_procedure.anchors_export import (
    AnchorCandidate,
    export_anchor_candidates,
)
from orchestrata_ultimate.factory.dor_procedure.models import ConstructionArtifact
from orchestrata_ultimate.factory.dor_procedure.validate import gaps, validate_artifact
from orchestrata_ultimate.factory.formalization.c4 import run_c4
from orchestrata_ultimate.factory.formalization.dor_bridge import c4_blocking_findings
from orchestrata_ultimate.factory.formalization.models import Formalization

BlockingSource = Literal["shape", "c3", "c4", "assumption", "residual", "anchor"]


@dataclass(frozen=True)
class DecisionInputs:
    """Everything :func:`run_decision_phase` needs about one decision run.

    ``artifact_path`` is optional provenance (the path the caller loaded
    ``artifact`` from, if any) — recorded in :attr:`DecisionRecord.provenance`
    only; an in-memory-only artifact leaves it empty.
    """

    requirement_ids: list[str]
    spec_text: str
    artifact: ConstructionArtifact
    artifact_path: str = ""


@dataclass(frozen=True)
class DetectorEngines:
    """Bundle of C3/C4 detector configuration for :func:`run_decision_phase`.

    ``rules``/``disabled``/``blacklist``/``repo_root`` configure the C3
    lexical/structural engine
    (:func:`~orchestrata_ultimate.factory.detectors.engine.run_detectors`);
    ``rules=None`` uses the full :data:`~...detectors.rules.ALL_RULES`
    registry. ``formalization`` is the optional declared formalization for
    C4 (:func:`~orchestrata_ultimate.factory.formalization.c4.run_c4`);
    ``None`` means no formalization has been authored for this requirement
    set, so C4 contributes no findings — C4 speaks only about the DECLARED
    formalization (ADR-0011's boundary), it is never inferred from prose.
    """

    rules: dict[str, tuple[RuleMeta, RuleFn]] | None = None
    disabled: AbstractSet[str] = frozenset()
    blacklist: Blacklist | None = None
    repo_root: Path | None = None
    formalization: Formalization | None = None


@dataclass(frozen=True)
class BlockingItem:
    """One named, human-readable blocking defect.

    ``source`` is one of the six gate sources (ADR-0015/0028); ``id`` is a
    stable identifier for the defect (an assumption id, a rule fingerprint, a
    requirement id, ...); ``statement`` says exactly what is wrong — no soft
    language.
    """

    source: BlockingSource
    id: str
    statement: str


@dataclass(frozen=True)
class RatificationItem:
    """One item queued for human ratification.

    ``kind`` is ``"anchor_proposal"`` (an unratified candidate anchor from
    :mod:`~orchestrata_ultimate.factory.dor_procedure.anchors_export`) or
    ``"assumption_escalation"`` (a B13 ledger assumption the arbiter cascade
    could not resolve automatically — :attr:`~...assumptions.models.
    AssumptionStatus.NEEDS_HUMAN`).
    """

    kind: Literal["anchor_proposal", "assumption_escalation"]
    requirement_id: str
    statement: str


@dataclass(frozen=True)
class DecisionRecord:
    """The result of one :func:`run_decision_phase` call.

    ``summary`` reads exactly ``"no blocking defects found"`` when
    ``verdict == "ready"`` — ADR-0015's wording is load-bearing: DoR can
    *refute* readiness, never *certify* completeness. Every field here is a
    plain JSON-able type (str/list/dict), so the record round-trips through
    ``dataclasses.asdict`` + ``json.dumps`` without a custom encoder.
    """

    verdict: Literal["ready", "blocked"]
    blocking: list[BlockingItem]
    ratification_queue: list[RatificationItem]
    provenance: dict[str, object]
    summary: str


def _shape_blocking(artifact: ConstructionArtifact) -> list[BlockingItem]:
    return [
        BlockingItem(source="shape", id=f"step{v.step}:{v.path}", statement=v.statement)
        for v in validate_artifact(artifact)
    ]


def _c4_blocking(formalization: Formalization | None) -> list[BlockingItem]:
    """C4 (B26): model-free logical detectors over a DECLARED formalization.

    Empty when no formalization was supplied — C4 has nothing declared to
    check, which is not the same as the requirement being contradiction-free
    or fully pinned (ADR-0011's boundary).
    """
    if formalization is None:
        return []
    report = run_c4(formalization)
    lines = c4_blocking_findings(report)
    items: list[BlockingItem] = []
    line_index = 0
    if report.contradiction is not None:
        items.append(
            BlockingItem(
                source="c4",
                id=f"{formalization.requirement_id}:contradiction",
                statement=lines[line_index],
            )
        )
        line_index += 1
    if report.under_constraint is not None:
        items.append(
            BlockingItem(
                source="c4",
                id=f"{formalization.requirement_id}:under_constraint",
                statement=lines[line_index],
            )
        )
    return items


def _c3_blocking(
    spec_text: str, source_name: str, engines: DetectorEngines
) -> list[BlockingItem]:
    """C3 (B22): lexical/structural detectors, Tier-1 (blocking) only."""
    rules = engines.rules if engines.rules is not None else ALL_RULES
    report = run_detectors(
        spec_text,
        source_name,
        rules=rules,
        disabled=engines.disabled,
        blacklist=engines.blacklist,
        repo_root=engines.repo_root,
    )
    return [
        BlockingItem(source="c3", id=finding.fingerprint, statement=finding.message)
        for finding in tier1_gate(report)
    ]


def _assumption_blocking(
    requirement_ids: list[str], ledger: AssumptionLedger
) -> tuple[list[BlockingItem], list[RatificationItem]]:
    """B13 assumption residue: any non-PINNED assumption for these
    requirements blocks; NEEDS_HUMAN ones are also queued for ratification.

    Mirrors :func:`~orchestrata_ultimate.factory.assumptions.dor.check_dor`'s
    "ready iff every assumption is PINNED" rule (AC-033-C1-02), scoped to
    ``requirement_ids`` rather than the whole ledger.
    """
    blocking: list[BlockingItem] = []
    queue: list[RatificationItem] = []
    wanted = set(requirement_ids)
    for assumption in ledger.assumptions():
        if assumption.requirement_id not in wanted:
            continue
        if assumption.status == AssumptionStatus.PINNED:
            continue
        blocking.append(
            BlockingItem(
                source="assumption",
                id=assumption.id,
                statement=(
                    f"assumption {assumption.id} at design node "
                    f"{assumption.design_node!r} is {assumption.status.value}, "
                    f"not pinned"
                ),
            )
        )
        if assumption.status == AssumptionStatus.NEEDS_HUMAN:
            queue.append(
                RatificationItem(
                    kind="assumption_escalation",
                    requirement_id=assumption.requirement_id or "",
                    statement=(
                        f"assumption {assumption.id} at design node "
                        f"{assumption.design_node!r} was escalated to human "
                        f"adjudication by the arbiter cascade"
                    ),
                )
            )
    return blocking, queue


def _residual_blocking(artifact: ConstructionArtifact) -> list[BlockingItem]:
    """Step-7 residual red cards: an unresolved GAP answer blocks."""
    return [
        BlockingItem(
            source="residual",
            id=f"{artifact.requirement_id}:{gap.axis_id}",
            statement=(
                f"residual axis {gap.axis_id!r} is an unresolved GAP for "
                f"requirement {artifact.requirement_id} (Example Mapping's "
                f"red card — an unresolved question blocks readiness)"
            ),
        )
        for gap in gaps(artifact)
    ]


def _anchor_blocking(
    requirement_ids: list[str],
    artifact: ConstructionArtifact,
    registry: AnchorRegistry,
) -> tuple[list[BlockingItem], list[RatificationItem]]:
    """ADR-0015's positive human-ratified floor: a requirement with no
    RATIFIED anchor blocks, and its DoR-derived candidates are queued.
    """
    blocking: list[BlockingItem] = []
    queue: list[RatificationItem] = []
    candidates_by_requirement: dict[str, list[AnchorCandidate]] = {}
    for candidate in export_anchor_candidates(artifact):
        candidates_by_requirement.setdefault(candidate.requirement_id, []).append(
            candidate
        )

    for requirement_id in requirement_ids:
        ratified = registry.anchors(
            requirement_id=requirement_id, status=AnchorStatus.RATIFIED
        )
        if ratified:
            continue
        blocking.append(
            BlockingItem(
                source="anchor",
                id=requirement_id,
                statement=(
                    f"requirement {requirement_id} has no ratified anchor "
                    f"(ADR-0015's positive human-ratified floor)"
                ),
            )
        )
        queue.extend(
            RatificationItem(
                kind="anchor_proposal",
                requirement_id=requirement_id,
                statement=(
                    f"anchor proposal for {requirement_id} "
                    f"({candidate.kind}, {candidate.input_name}:"
                    f"{candidate.class_label}): given {candidate.given!r} "
                    f"expect {candidate.expect!r}"
                ),
            )
            for candidate in candidates_by_requirement.get(requirement_id, [])
        )
    return blocking, queue


@traces(spec="033", reqs=["REQ-033-C5"])
def run_decision_phase(
    inputs: DecisionInputs,
    *,
    ledger: AssumptionLedger,
    anchor_registry: AnchorRegistry,
    detector_engines: DetectorEngines,
) -> DecisionRecord:
    """Run the ADR-0015/0028 decision phase and return its verdict.

    Pure orchestration: no LLM call, deterministic given ``inputs`` and the
    state of ``ledger``/``anchor_registry``. Order matters — a shape-invalid
    artifact reports ONLY shape blocks; nothing downstream is scored (see
    the module docstring for the full six-stage order and rationale).
    """
    timings: dict[str, float] = {}
    engines_ran: list[str] = []

    start = time.perf_counter()
    shape_blocking = _shape_blocking(inputs.artifact)
    timings["shape"] = time.perf_counter() - start
    engines_ran.append("shape")

    if shape_blocking:
        return DecisionRecord(
            verdict="blocked",
            blocking=shape_blocking,
            ratification_queue=[],
            provenance={
                "engines_ran": engines_ran,
                "artifact_path": inputs.artifact_path,
                "timings_seconds": timings,
            },
            summary=f"{len(shape_blocking)} blocking defect(s) found",
        )

    blocking: list[BlockingItem] = []
    ratification_queue: list[RatificationItem] = []

    start = time.perf_counter()
    blocking.extend(_c4_blocking(detector_engines.formalization))
    timings["c4"] = time.perf_counter() - start
    engines_ran.append("c4")

    start = time.perf_counter()
    source_name = f"requirements:{','.join(inputs.requirement_ids)}"
    blocking.extend(_c3_blocking(inputs.spec_text, source_name, detector_engines))
    timings["c3"] = time.perf_counter() - start
    engines_ran.append("c3")

    start = time.perf_counter()
    assumption_blocking, assumption_queue = _assumption_blocking(
        inputs.requirement_ids, ledger
    )
    blocking.extend(assumption_blocking)
    ratification_queue.extend(assumption_queue)
    timings["assumption"] = time.perf_counter() - start
    engines_ran.append("assumption")

    start = time.perf_counter()
    blocking.extend(_residual_blocking(inputs.artifact))
    timings["residual"] = time.perf_counter() - start
    engines_ran.append("residual")

    start = time.perf_counter()
    anchor_blocking, anchor_queue = _anchor_blocking(
        inputs.requirement_ids, inputs.artifact, anchor_registry
    )
    blocking.extend(anchor_blocking)
    ratification_queue.extend(anchor_queue)
    timings["anchor"] = time.perf_counter() - start
    engines_ran.append("anchor")

    # B14/C2 seam: interpretation-mutation findings plug in here as a further
    # detector stage once built — same BlockingItem contract.

    verdict: Literal["ready", "blocked"] = "blocked" if blocking else "ready"
    summary = (
        "no blocking defects found"
        if verdict == "ready"
        else f"{len(blocking)} blocking defect(s) found"
    )
    return DecisionRecord(
        verdict=verdict,
        blocking=blocking,
        ratification_queue=ratification_queue,
        provenance={
            "engines_ran": engines_ran,
            "artifact_path": inputs.artifact_path,
            "timings_seconds": timings,
        },
        summary=summary,
    )