orchestrata_ultimate.factory.decision.authority

Reference › orchestrata_ultimate › factory › decision › authority

authority is the design authority of the decision phase (SPEC-033 row B8, ADR-0028, requirement REQ-033-C5). ADR-0028 delivers the Definition-of-Ready axes as a construction procedure rather than a checklist, because a construction step either produces its output or visibly cannot. The design authority is the component that runs that procedure from dor_procedure. DesignAuthority is the protocol the decision phase depends on, with one async method, construct(requirement_ids, spec_text), returning a ConstructionArtifact. LLMDesignAuthority is the live implementation: it makes one LLM call through llm.client, using a system prompt that lists the procedure steps in order, and parses the reply into an artifact. It routes through the agent alias DESIGN_AUTHORITY_VOLK (design_authority), which the code comments tie to the strongest configured Claude model in every profile. A real async function may be injected as _call_llm for tests, the project's only sanctioned test double. The scripts.factory_decision CLI imports the module; tests are tests/factory/test_design_authority.py and tests/factory/test_design_authority_live.py.

At a glance

Class diagram

Dependency diagram

API

Classes

DesignAuthority (protocol)

Bases: Protocol · Line 107

ADR-0028's constructive reviewer: RUNS the B23 construction procedure.

LLMDesignAuthority

Decorators: traces(spec='033', reqs=['REQ-033-C5']) · Line 126

Live :class:DesignAuthority: one LLM call, parsed via :func:~orchestrata_ultimate.factory.dor_procedure.artifact. _artifact_from_dict — the artifact module's own from-dict path, so parsing stays…

Source

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

"""The design authority (ADR-0028's constructive reviewer, B8).

ADR-0028: DoR's completeness axes are delivered as a construction procedure
(B23) rather than a checklist, because a model handed a flat checklist
produces a plausible tick per row while a construction step either produces
its output or visibly cannot. The design authority is the thing that RUNS
that procedure — :class:`DesignAuthority` is the seam
:mod:`~orchestrata_ultimate.factory.decision.phase` depends on;
:class:`LLMDesignAuthority` is its live, LLM-backed implementation.
"""

from __future__ import annotations

import json
from collections.abc import Awaitable, Callable
from typing import Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.dor_procedure.artifact import _artifact_from_dict
from orchestrata_ultimate.factory.dor_procedure.models import ConstructionArtifact
from orchestrata_ultimate.factory.dor_procedure.rulings import list_ruling_ids
from orchestrata_ultimate.llm.client import call_llm

# The sole sanctioned LLM test double for this project (see AGENTS.md):
# inject a real async function as ``_call_llm`` -- never mock/patch
# ``llm.client.call_llm`` itself.
_CallLLMFn = Callable[..., Awaitable[str]]

# The llm/config.py agent alias (configs/spike-profiles.yaml entry) the
# design authority's calls route through -- provider anthropic, the
# strongest configured claude model, in every profile (the constructive
# reviewer is never downgraded to a cheaper active profile's default model).
DESIGN_AUTHORITY_VOLK = "design_authority"

_SYSTEM_PROMPT = """You are the design authority for a software requirement, \
under the SPEC-033 governed autonomous build factory. Your job is ADR-0028's \
"constructive reviewer": RUN the B23 DoR construction procedure (Perspective- \
Based Reading, not a checklist) over the given requirement(s) and spec text, \
and produce ONE JSON object -- a ConstructionArtifact. A construction step \
either produces its output or visibly cannot; never tick a step without \
doing the construction it names.

Carry out the procedure in this exact order -- each step forces a subset of \
the completeness axes:

1. Enumerate the inputs of the criterion. Are they all specified? Are any \
specified but unused?
2. Build the equivalence classes and boundaries for each input -- at least \
one VALID class and either two invalid classes (INVALID_BELOW + \
INVALID_ABOVE) or, for set-membership inputs, a COMPLEMENT class.
3. Write the expected behaviour for every class from step 2, including \
every invalid class.
4. Walk the state machine -- every state x every input, plus the no-input \
case.
5. Walk the failure envelope for every dependency -- down, slow, partial, \
retry, restart. If there truly are no dependencies, say so explicitly \
(a stated reason, never a silent omission).
6. Walk the limits -- rate, size, count, time (N/A-with-reason allowed).
7. Answer EVERY one of these residual axes as forced choice -- SPECIFIED \
(with a "where" citation), NOT_APPLICABLE (with a reason), or GAP (an \
honest unresolved question, never silence): measurable-threshold, \
authorization-default, feedback-loop, observability, \
omitted-non-functionals, time-timezone, encoding-locale, \
domain-ambiguity-probe, external-completeness, divergent-interpretation-probe.

Then, LAST and separately from the seven numbered steps above, fill the \
rulings checklist (workflow-improvements-plan.md Step 6.3): name every \
standing project ruling the work dispatched for this requirement will have \
to obey, with the concrete reason it applies. Ruling ids look like \
"RUL-001".."RUL-020" and are listed for you in the user message; use ONLY \
ids from that list. The point is to catch a ruling violation at DISPATCH \
time rather than reject the finished work. If genuinely none apply, say so \
in "none_apply_reason" -- silence is not an allowed answer, and an empty \
checklist with no reason BLOCKS the artifact.

Respond with ONLY a single JSON object matching this exact shape -- no \
markdown fences, no prose before or after:

{
  "requirement_id": "<the requirement id>",
  "author_identity": "design_authority",
  "created_at": "<ISO-8601 UTC timestamp>",
  "schema_version": 1,
  "step1": {"inputs": [{"name": "<str>", "specified": <bool>, "where": \
"<str>", "used": <bool>}], "unused_inputs": ["<str>"]},
  "step2": {"per_input": {"<input name>": [{"label": "<str>", "kind": \
"valid|invalid_below|invalid_above|complement", "description": "<str>", \
"boundary": "<str>"}]}},
  "step3": {"behaviors": {"<input>:<class label>": "<expected behaviour>"}},
  "step4": {"states": ["<str>"], "inputs": ["<str>"], "transitions": \
{"<state>|<input>": "<behaviour or 'N/A: <reason>'>"}, "no_input_case": \
"<str>"},
  "step5": {"dependencies": ["<str>"], "responses": \
{"<dependency>|down|slow|partial|retry|restart": "<response or 'N/A: \
<reason>'>"}, "no_dependencies_reason": "<str>"},
  "step6": {"limits": {"rate": "<str>", "size": "<str>", "count": "<str>", \
"time": "<str>"}},
  "step7": {"answers": [{"axis_id": "<str>", "kind": \
"specified|not_applicable|gap", "where": "<str>", "reason": "<str>"}]},
  "rulings_checklist": {"entries": [{"ruling_id": "<RUL-NNN>", \
"applies_because": "<why this ruling binds THIS work>"}], \
"none_apply_reason": "<str, only when entries is empty>"}
}
"""


class DesignAuthority(Protocol):
    """ADR-0028's constructive reviewer: RUNS the B23 construction procedure.

    ``construct`` is the sole seam — it takes the requirement set and the
    spec text those requirements are drawn from, and returns the resulting
    :class:`~orchestrata_ultimate.factory.dor_procedure.models.
    ConstructionArtifact`. The Protocol itself implies no LLM call — a test
    double that builds a :class:`ConstructionArtifact` directly satisfies it
    just as well as :class:`LLMDesignAuthority`.
    """

    async def construct(
        self, requirement_ids: list[str], spec_text: str
    ) -> ConstructionArtifact:
        """Run the B23 procedure over ``spec_text`` and return the artifact."""
        ...


@traces(spec="033", reqs=["REQ-033-C5"])
class LLMDesignAuthority:
    """Live :class:`DesignAuthority`: one LLM call, parsed via
    :func:`~orchestrata_ultimate.factory.dor_procedure.artifact.
    _artifact_from_dict` — the artifact module's own from-dict path, so
    parsing stays in exactly one place.

    Routes through :func:`orchestrata_ultimate.llm.client.call_llm` with
    ``volk=DESIGN_AUTHORITY_VOLK`` and ``temperature=0.0`` (the construction
    procedure is meant to be as reproducible as an LLM call can be);
    ``model=None`` so the active ``configs/spike-profiles.yaml`` profile
    resolves provider/model (anthropic, the strongest configured claude
    model, per every profile's ``design_authority`` entry).

    A parse failure — malformed JSON, or JSON that does not match
    :class:`~orchestrata_ultimate.factory.dor_procedure.models.
    ConstructionArtifact`'s from-dict contract — raises. This class never
    fabricates a partial artifact.

    ``_call_llm`` is an optional injection seam for tests (the project's
    sole sanctioned LLM test double, per AGENTS.md): pass a real async
    function with :func:`~orchestrata_ultimate.llm.client.call_llm`'s
    signature to exercise the parsing/error paths without a network call;
    it defaults to the real ``call_llm``.
    """

    def __init__(
        self,
        *,
        volk: str = DESIGN_AUTHORITY_VOLK,
        _call_llm: _CallLLMFn | None = None,
    ) -> None:
        self._volk = volk
        self._call_llm = _call_llm if _call_llm is not None else call_llm

    async def construct(
        self, requirement_ids: list[str], spec_text: str
    ) -> ConstructionArtifact:
        """Call the LLM once and parse its response into a ConstructionArtifact.

        Raises:
            ValueError: If the response is not valid JSON, is not a JSON
                object, or the object does not match ``ConstructionArtifact``'s
                expected shape (missing/invalid fields, unsupported
                ``schema_version``).
        """
        # Step 6.3: the model can only cite ruling ids it has been shown, and
        # `validate_artifact` rejects any id that is not in the canon — so the
        # list is derived from the canon here, never hardcoded in the prompt
        # (a hardcoded list would silently rot the moment a ruling is added).
        ruling_ids = list_ruling_ids()
        rulings_line = (
            f"Known ruling ids: {', '.join(ruling_ids)}"
            if ruling_ids
            else (
                "Known ruling ids: (canon unavailable — leave entries empty "
                "and state that in none_apply_reason)"
            )
        )
        user_message = (
            f"Requirement id(s): {', '.join(requirement_ids)}\n\n"
            f"{rulings_line}\n\n"
            f"Spec text:\n{spec_text}"
        )
        raw = await self._call_llm(
            system_prompt=_SYSTEM_PROMPT,
            user_message=user_message,
            temperature=0.0,
            volk=self._volk,
        )
        try:
            data = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise ValueError(
                f"design authority response is not valid JSON: {exc}\n"
                f"response was: {raw!r}"
            ) from exc
        if not isinstance(data, dict):
            raise ValueError(
                f"design authority response must be a JSON object, got "
                f"{type(data).__name__}"
            )
        return _artifact_from_dict(data)